Doris的执行计划生成、分发与执行

目录

一、概述

三、执行计划的生成

四、执行计划的分发

五、执行计划的执行

六、关于PipeLine

七、Stream Load 的执行计划

八、举个例子

一、概述

执行SQL的代码入口为StmtExecutor::execute()

三、执行计划的生成

在Doris的FE端，与大多数数据库系统一样，要从SQL或某种http请求，生成执行计划，从SQL生成，一开始是“抽象语法树”（Abstract Syntax Tree），这个抽象语法树不一定是规则的二叉树，而只是一些语法对象，通过类的成员变量联系起来，例如：

fe的语法定义文件为 ./fe-core/src/main/cup/sql_parser.cup ，这也是分析代码开始的地方，从这里可以看到一个命令被转换为怎样一个数据结构（AST）。

然后经过分析，重写步骤（有些数据库称为bind，语义分析等等），将AST改造成逻辑语法树，这个逻辑语法树，就是以关系运算符为主干的二叉树了，或者近似于这样一个二叉树了。

这个称为逻辑计划，生成逻辑计划以后，就可以进行各种优化了。在java代码中，逻辑计划的节点都是PlanNode的子类的对象，AST的节点都有个analyze()，analyze()被层层调用，生成逻辑计划树。

数据是分布式的分散到BE的，执行计划也是在BE节点上执行，而不是在FE节点上执行，FE只负责生成执行计划，并决定把执行计划发给谁。

接下来就是生成物理执行计划，并且把执行计划分布式化。

就是把PlanNode组成的执行计划树，分割成不同部分，这些不同部分称为fragment，代码中用PlanFragment类的对象表示，这些不同部分可以在不同的BE节点上执行，并且在BE节点上执行的同样一个PlanFragment，可以有多个并行执行的示例，虽然每个实例是一样的操作逻辑，但是读取的是同一个表的不同部分，这些部分称为ScanRange，例如一个PlanFragment的两个实例，它们在同一个BE节点上，读取同一个表dup1，它们的Scan操作符读取的也是同一个表，但是不同的实例的Scan操作符，读取的是不同的tablet，每个Scan操作符有自己的独一无二的ScanRange，ScanRange是一个tablet列表。

参考FE代码：Coordinator::computeScanRangeAssignment()

调试时，可以调用这个函数 tablets_id_to_string(_scan_ranges) 返回ScanRange里的tablet_id。
//every doris_scan_range is related with one tablet so that one olap scan node contains multiple tablet

Fragment之间通过网络通讯，新增了两个算子专门联系两个Fragment之间的运算符，就是DataSink和ExchangeNode，例如上图，之前直接联系的Hash Join Node和OlapScanNode，在分到不同的Fragment后通过DataSink和ExchangeNode沟通数据。

FE会决定执行计划划分为几个Fragment，并且决定这些Fragment分发到哪个BE上执行，也决定分发到BE上的Fragment，要创建几个实例，这些实例的scan操作符的ScanRange是什么，总之BE只负责无脑执行，所有执行细节都有FE在创建最终执行计划时设置好了。

过程调用的thrift和protobuf代码在doris/gensrc目录下，编译过程中，会根据其中的定义生成java和cpp文件。

fe使用生成的java文件，还需要将这些java文件复制到fe/fe-common/src/main/java/org/apache/doris/thrift目录下，然后编译，可见fe-be之间的通讯应该是走thrift RPC，be-be之间的通讯应该会走protobuf。

be使用生成的cpp文件，不必将这些cpp文件复制到be目录，而是直接编译。

四、执行计划的分发

执行计划在FE上生成完毕，由FE直接下发给需要执行的它的BE，而不会是先下发给一个所谓的coordinator BE，然后又它再分给其它BE，注意这一点，容易引起误会，select这样的计划，最终数据会汇总到一个BE上，再由这个BE传给FE，这个BE称为Root BE，它负责执行时数据的最终汇总，但是不负责执行计划的分发！

FE下发执行计划的入口函数是：

Coordinator::exec()

      |__> Coordinator::sendPipelineCtx()
底层调BackendServiceClient::execPlanFragmentPrepareAsync()，通过grpc把fragments信息发给BE。

在FE的代码里，PlanFragment里的planRoot成员变量，指向自己所包含的执行计划片段的最上层的一个节点，每个执行计划算子（PlanNode）都有一个fragment成员变量，指向自己所在的PlanFragment对象。

PlanFragment的children成员变量，将父子fragment联系起来。

在Coordinator::sendPipelineCtx() 中，beToPipelineExecCtxs存的是发给所有BE的fragment信息，其中每个PipelineExecCtxs是发给同一个BE的所有fragment信息（可能向一个BE发送多个不同的fragment，并且同一个fragment会有多个实例），一个PipelineExecCtxs包含多个PipelineExecContexts，一个PipelineExecContexts对应一个fragment，一个PipelineExecContexts，对象里又可能包含多个PipelineExecContext，每个表示一个fragment实例。

 FE中Coordinator这个类很重要，里面有个fragment list，就是要发给BE的要执行的fragment。

图片来自（Doris 源码分析 (五) gRpc 与 thrift 接口 - 简书 (jianshu.com)）

FE是通过gRPC向BE分发Fragments的。

FE与BE之间的RPC调用，是有超时的，在FE端fe.conf通过下面两个参数可以设置超时时间：

backend_rpc_timeout_ms

remote_fragment_exec_timeout_ms

FE向BE分发fragment，并不是每个BE都分发相同的fragment，而其中发给Root BE的fragment与其它BE稍有不同，多了顶部的fragment，其它BE的数据汇总到这个Root BE，然后从这个Root BE统一发给FE，注意，不是每个BE分别向FE发数据。

BE之间的数据传输，底层也是用grpc。最底层调用

doris::PBackendService_Stub::transmit_block()

关于执行计划（fragments）的分发，是从FE直接向需要执行执行计划的BE发fragments，而不是发给coordinator BE由它转发给其它BE。

笼统的说，FE向BE分发执行计划并执行，大体分两种情况：

1、如果执行计划只有一个fragment，那么FE只向BE发一个RPC（BackendServiceClient::execPlanFragmentAsync()），把执行计划发给BE，BE端根据信息重建ExecNode组成的执行计划树，并且执行。注意，不管哪种情况，fragment信息通过rpc到达BE后，其中plan都有一个reconstruct的过程！

2、如果执行计划中有多个fragment，会分两步，第一步是FE调用BackendServiceClient::execPlanFragmentPrepareAsync()下发fragment，在BE端响应了这个RPC后，会根据fragment信息，重建ExecNode组成的执行计划树，但是不执行，当把所有fragment的执行计划树都重建好了，即prepare完毕。然后FE端再调用BackendServiceClient::execPlanFragmentStartAsync()，让BE上刚才准备好的执行计划开始执行。

上述逻辑FE端的代码在 Coordinator::sendPipelineCtx()

BackendServiceClient::execPlanFragmentPrepareAsync

BackendServiceClient::execPlanFragmentStartAsync

FE这两个函数已经比较底层了，里面就调用最底层的stub。

BE、FE交互的许多类型，定义在doris/gensrc/build/gen_cpp/下生成的文件里。

五、执行计划的执行

FE调用BackendServiceClient::execPlanFragmentPrepareAsync
导致BE调用PInternalServiceImpl::exec_plan_fragment_prepare
在这里面request参数包括所有fragment的信息。
(在我的单机FE+BE各一个环境，是FE一次远程调用，向BE下发所有fragment，
放在PExecPlanFragmentRequest->request里,这是个字符串,需要进行反序列化)
一步一步往下调用，在PInternalServiceImpl::_exec_plan_fragment里，
从FE传来的所有fragment信息，被反序列化到TPipelineFragmentParamsList，
里面每个param是一个fragment信息，每个fragment调用一次fragment_mgr()->exec_plan_fragment()，
进而调用PipelineFragmentContext::prepare(),以ExecNode的子类为节点构造执行计划树
PipelineFragmentContext::prepare->ExecNode::create_tree()。

FE调用BackendServiceClient::execPlanFragmentStartAsync
导致BE调用PInternalServiceImpl::exec_plan_fragment_start
进一步调用FragmentMgr::start_query_execution()（这个函数，整个query只调一次，不是每个fragment调一次）
设置query_id所指的执行计划为可执行状态

在BE的PInternalServiceImpl::_exec_plan_fragment()中，通过RPC 传来的参数TPipelineFragmentParams，代表一个fragment，其中的local_params，每个元素代表这个fragment的instance，每个元素的类型是TPipelineInstanceParams。它们的定义在 gensrc/thrift/PaloInternalService.thrift。

在BE端，谁分配到了最顶层的fragment，谁就是这次查询的Root Fragment。不同BE间，sink和exchange的通讯，是基于brpc（应该是百度内部优化过的rpc），BE代码中相关函数：transmit_block/transmit_data。

sink的底层是VDataStreamSender，内部的_channels数组，是Channel对象或PipChannel对象数组，一个Channel表示发给一个上层的exchange节点的通道，例如，OlapScanNode的VDataStreamSender有6个Channel对象，表示从tablet扫描到的记录，按照hash函数，分发给6个上层的Exchange节点。关于如何根据FE下发的信息创建VDataStreamSender，参考DataSink::create_data_sink()。

Channel底层是RPC调用，使用gRPC（百度版本的RPC？），接口定义在gen_cpp/internal_service.pb.cc/h里定义的PBackendService，在internal_service.proto里定义：

Channel::init()

Channel/PipChannel::add_rows() -- 积累行记录

Channel/PipChannel::send_(local_)block() -- 向另一个fragment发送行记录

在BE上fragment prepare（包括重建，准备好各种数据收发对象）完成后，就开始执行了，新的执行引擎模型称为pipeline，它与火山模型不同的是，不是通过遍历执行计划树来执行的，而是再把每个算子或fragment分成若干个operator，operator之间可以并行执行。PipelineTask是被pipeline系统调用的对象，可以理解为线程。整个pipeline引擎类似于一个线程池，一个BE只有一个pipeline引擎（TaskScheduler对象），和线程池不同的是，pipeline的线程在遇到阻塞时，会放弃任务，然后去执行其它不会阻塞的任务。pipeline引擎的阻塞任务队列有一个，就绪任务队列有好几个，有一个专门线程不断检查阻塞任务队列，将其中不再阻塞的任务（PipelineTask对象），加到其中一个就绪队列中，有好几个线程会从就绪队列里取PipelineTask执行。

在整个BE中，有一个（也只有一个）ExecEnv对象，通过ExecEnv::GetInstance()获得，里面包含了TaskScheduler对象（全局只有一个），这就是流水线的执行对象，TaskScheduler对象里包含了BlockedTaskScheduler对象。BE中还有一些其它模块的线程池，BE中有可能每个模块都有自己的线程池。

相关代码在：

ExecEnv::init_pipeline_task_scheduler()和TaskScheduler::start()

BE中Fragment、Pipeline、PipelineTask、Operator对象的关系：一个PipelineTask对应一个Pipeline对象，一个PipelineTask处理多个operator，operator chain在PipelineTask::_operators成员中。PipelineTask的operator chain是从Pipeline创建来的--Pipeline::build_operators()。

operator可以对应一个ExecNode，也可能一个ExecNode对应多个Operator（参考PipelineFragmentContext::_build_pipelines()中的逻辑就明白了，列入HashJoinNode和它的子节点会变成两个pipeline）。Operator的执行最终还是要调用相应ExecNode里的算法函数执行，一个Fragment会被分成多个Pipeline，一个Pipeline里有多个operator由一个PipelineTask执行。使用operator后，ExecNode的get_next不用了，用push和pull。

PipeLineTask对象和PipeLine对象是一一对应的，PipeLine定义了所有静态信息，如一个pipeline中有哪些operator（operator 链），sink operator，这些operator中会有指针指向对应的ExecNode，毕竟执行代码在ExecNode的方法里。

VMysqlResultWriter从BE将结果写给FE，FE通过BackendServiceClient::fetchDataAsync 接收BE的 VResultSink::send发来的数据。

BE中属于一个Fragment的ExecNode，创建出来的一些列的pipeline，归一个PipelineFragmentContext管理，pipeline由PipelineFragmentContext::_build_pipelines和_build_pipeline_tasks创建，最终由 PipelineFragmentContext::submit提交给pipeline执行引擎。

六、关于PipeLine

pipeline是doris里非常关键的执行引擎，查询、streamload等等操作最终都会以pipeline的方式执行，所以不可以不对它深入理解，最近看代码，记录一下对pipeline机制、代码的理解。

执行计划从FE发送到BE后，从Thrift结构体转为 ExecNode 对象树，是在 ExecNode::create_tree 做的。之后可以通过遍历树执行（火山模型），也可以将这个执行计划转为若干pipeline，由pipeline引擎执行。

为了更好的理解pipeline代码和它如何工作的，避免理解时的困惑，有必要了解的一些信息：

1、pipeline引擎底层是基于ThreadPool（见threadpool.h），这个ThreadPool类是BE中广泛使用的，每个模块都有一个或多个属于自己的ThreadPool，管理若干用于自己模块的thread，使用线程池的关键，就是把你要执行的函数告诉线程池。

2、要了解BE中的代码，模块与模块之间，大多数情况下，并不是在一个线程中调用的，而是模块A产生数据放到一个缓存或队列里，模块B在另一个线程中读取这个数据。这样，查看函数调用栈来理解代码，就不理想了。应该更多的关注文档、中间数据结构（那个队列或缓存，数据流转的统一格式，哪个函数读写这个数据），并且不必追踪到太底层，一般跟踪到读取和写入缓存或队列的函数即可，再适当关心一下统一的数据格式。

一个fragment的所有ExecNode所对应的pipeline由一个PipelineFragmentContext管理，其中可能包含多个pipeline，一个pipeline又可能包含多个operator。

将ExecNode Tree转为pipeline的函数是 PipelineFragmentContext::_build_pipelines，ExecNode和pipeline、operator如何对应，都在这个函数里决定，它会被递归调用，遍历ExecNode Tree。ExecNode会对应一个或多个operator，一个pipeline对象会包含来自一个或多个ExecNode的operator。

例如 HashJoinNode会产生出HashJoinProbeOp和HashJoinBuildSinkOp两个operator，它对应的pipeline会包含HashJoinProbeOp，和它的子节点VNewOlapScanNode产生的ScanOperator，而HashJoinBuildSinkOp是另一个pipeline的_sink成员，这个pipeline的operator链中只包含一个ExechangeSourceOperator，但这个HashJoinBuildSinkOp会引用HashJoinNode并且调用它的sink，注意，ExecNode的sink方法其实是外界向本节点输入数据的调用函数，是被外界的operator或ExecNode调用的，不是本ExecNode向外输出的数据的方法，从生命中input_block的参数名可以看出。

pipeline的_sink成员指向的operator不一定是DataSinkOperator的子类，更有可能是StreamingOperator的子类。

一个pipeline中比较重要的是operator链，还有一个_sink成员，pipeline执行时（PipelineTask::execute），调用operator的get_block，从而调用operator链上所有operator的get_block，这些operator再调用它们对应的ExecNode的pull方法。调用get_block获得数据（block）后再调用sink将数据发给下游pipeline。

pipeline的operator有四种：

DataSinkOperator -- 对应SinkNode的子类，关于SinkNode对应的operator的创建在PipelineFragmentContext::_create_sink

StreamingOperator 

SourceOperator -- StreamingOperator 的子类

StatefulOperator -- StreamingOperator 的子类

它们的区别除了 DataSinkOperator 要指向 SinkNode 子类外，其它三个的区别在于get_block的逻辑不同，而各个运算符（ExecNode）的operator大多是继承自这四个operator。

每个operator都有个指向ExecNode的指针，也都有个对应的builder，虽然我觉得大多数builder没有实质性的代码，为每个ExecNode创建operator和builder有点不必要，也许直接用父类即可。

创建新ExecNode，并且让它能够被pipeline引擎执行，需要做什么？

1、需要为他创建operator builder和operator，一般继承自StreamingOperator 或 StatefulOperator。

2、然后具体的计算逻辑要实现 ExecNode::pull，ExecNode::_sink，ExecNode::alloc_resource，ExecNode::get_next_after_projects

七、Stream Load 的执行计划

stream load 是BE向FE请求执行计划，FE生成执行计划后，通过RPC发给BE，BE上重建执行计划，重建后可以直接执行，或者通过pipeline引擎执行，一下是直接执行的流程：

FE下发执行计划给BE后，就不管了，BE直接从http客户端取得数据，客户端会把数据分块传输给BE，每传一块BE的StreamLoadAction::on_chunk_data()就被调用一次，这个函数会把数据缓存给StreamLoadPipe，同时运行执行计划的执行线程也在执行，ScanNode 的 scanner 不断从 StreamLoadPipe 取数据发给下一个节点 VOlapTableSink， 它调用VNodeChannel根据分区和分桶，通过RPC把记录发给不同的BE在上面会执行存储引擎的写盘逻辑。（如何分发，是否要排序，这部分逻辑在哪里？在VOlapTableSink的方法里吗）

八、举个例子

集群结构为一台FE节点，三台BE节点，两个duplicate类型表：

CREATE TABLE IF NOT EXISTS dup1
(
    `user_id` BIGINT NOT NULL AUTO_INCREMENT COMMENT "用户id",
    `date` DATE NOT NULL COMMENT "数据灌入日期时间",
    `city` VARCHAR(20) COMMENT "用户所在城市",
    `age` SMALLINT COMMENT "用户年龄",
    `sex` TINYINT COMMENT "用户性别",
    `cost` BIGINT DEFAULT "0" COMMENT "用户总消费"
)
DUPLICATE KEY (`user_id`)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 16
PROPERTIES (
"replication_allocation" = "tag.location.default: 3"
);

CREATE TABLE IF NOT EXISTS dup2
(
    `user_id` BIGINT NOT NULL AUTO_INCREMENT COMMENT "用户id",
    `date` DATE NOT NULL COMMENT "数据灌入日期时间",
    `city` VARCHAR(20) COMMENT "用户所在城市",
    `age` SMALLINT COMMENT "用户年龄",
    `sex` TINYINT COMMENT "用户性别",
    `cost` BIGINT DEFAULT "0" COMMENT "用户总消费"
)
DUPLICATE KEY (`user_id`)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 16
PROPERTIES (
"replication_allocation" = "tag.location.default: 3"
);

查询SQL：

select count(1) from dup1 join dup2 on dup1.age = dup2.cost;

我们从FE生成逻辑Fragments开始分析，在上面这个环境中，FE为这条查询生成的Fragments（执行计划）如下：

发到BE节点后，根据FE发来的信息，重建执行计划树，但是在BE中没有PlanFragment这种对象，只有FE中的PlanNode，ExchangeNode，SinkNode在BE中有对应的对象创建。

对于这个例子，BE中的执行计划是这样的：

1、有三个BE节点，Fragment0里的算子，会分配到其中一个BE中，且只有这一个BE会有Fragment0，其他的Fragment的数据最终都会汇总到这个BE的Fragment0中，再发给FE，这个BE称为Root BE。

2、Fragment1里的算子，在每个BE上都会创建2个实例，整个BE集群里一共有6个join节点。

3、Fragment2、Fragment3里的算子，在每个BE上都会创建2个实例，整个BE集群里，读取dup1的ScanNode有6个，读取dup2的ScanNode有6个。

4、Fragment2、Fragment3里的每个ScanNode，都读取表的不同部分，每个ScanNode有一个ScanRange对象，其中的tablet id表示这个ScanNode要读取哪些tablet，每个ScanNode的ScanRange中的tablet id都不重复。

5、表dup1和表dup2，按照shuffle join，做分布式join，即读取dup1、dup2的每个ScanNode都按照相同的hash函数，将读取的记录分发给3个BE里的6个join节点（注意：每个ScanNode可能向所有6个join发记录，而不是只向其中3个join发记录，这6个join每个都是hash函数的全局不同的桶），如果两个表中有可join的行，那么它们一定被分发到同一join节点，这样就把两个大表的join工作分而治之了，每个join节点只管对发给自己的两个表的行做join，结果再发给上一层节点，即AggregateNode。

6、整个集群有6个join，也有6个join上层的AggregateNode，这些AggregateNode做一部工作，再把数据发给Root BE的AggregateNode完成汇总发给FE。

下图是BE中的算子以及数据流，scanNode和exchangeNode的数据流用send to EXCH_X+bucket{...}表示，对于每个scanNode是根据hash函数，把记录分发到全部6个JOIN的exchange节点的，这个分发是跨BE的，scanNode到一个exch节点数据流，对应一个VDataStreamSender的Channel。

参考：

Doris全面解析】Doris SQL 原理解析 (qq.com)

Doris原理分享(2) - 知乎 (zhihu.com)

Pipeline 执行引擎 - Apache Doris

Apache Doris 源码阅读与解析系列直播——第四讲 一条SQL的执行过程_哔哩哔哩_bilibili

Doris 源码分析 (五) gRpc 与 thrift 接口 - 简书 (jianshu.com)

Doris 源码分析 (三) 基础语法 - 简书 (jianshu.com)文章来源地址https://www.toymoban.com/news/detail-655818.html

到了这里，关于Doris的执行计划生成、分发与执行的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！