DuckDB核心模块揭秘 | 第1期 | 向量化执行引擎之Pipeline

DuckDB是一款非常火的OLAP嵌入式数据库，性能超级棒。它分为多个组件：解析器、逻辑规划器、优化器、物理规划器、执行器以及事务和存储管理层。其中解析器原语PgSQL的解析器；逻辑规划器包含binder、plan generator，前者解析所有引用的schema中的对象的表达式，将其与列名和类型匹配，后者将binder生成的AST转换成由基本逻辑查询运算符组成的树；优化器产生优化的查询计划；物理规划器将优化的查询计划转换成物理执行计划，即PhysicalOperator树。它的高性能主要得益于它的push-based pipeline向量化执行引擎。本文介绍下它的向量化引擎pipeline生成原理。

1、物理执行计划长什么样？有哪些算子？

physical_plan_generator.cpp中CreatePlan函数将逻辑计划节点转换成物理计划节点，即PhysicalOperator。有哪些算子类型呢？PhysicalOperatorType：

//===--------------------------------------------------------------------===//
// Physical Operator Types
//===--------------------------------------------------------------------===//
enum class PhysicalOperatorType : uint8_t {
  INVALID,
  ORDER_BY,
  LIMIT,
  STREAMING_LIMIT,
  LIMIT_PERCENT,
  TOP_N,
  WINDOW,
  UNNEST,
  UNGROUPED_AGGREGATE,
  HASH_GROUP_BY,
  PERFECT_HASH_GROUP_BY,
  FILTER,
  PROJECTION,
  COPY_TO_FILE,
  BATCH_COPY_TO_FILE,
  FIXED_BATCH_COPY_TO_FILE,
  RESERVOIR_SAMPLE,
  STREAMING_SAMPLE,
  STREAMING_WINDOW,
  PIVOT,


  // -----------------------------
  // Scans
  // -----------------------------
  TABLE_SCAN,
  DUMMY_SCAN,
  COLUMN_DATA_SCAN,
  CHUNK_SCAN,
  RECURSIVE_CTE_SCAN,
  CTE_SCAN,
  DELIM_SCAN,
  EXPRESSION_SCAN,
  POSITIONAL_SCAN,
  // -----------------------------
  // Joins
  // -----------------------------
  BLOCKWISE_NL_JOIN,
  NESTED_LOOP_JOIN,
  HASH_JOIN,
  CROSS_PRODUCT,
  PIECEWISE_MERGE_JOIN,
  IE_JOIN,
  DELIM_JOIN,
  INDEX_JOIN,
  POSITIONAL_JOIN,
  ASOF_JOIN,
  // -----------------------------
  // SetOps
  // -----------------------------
  UNION,
  RECURSIVE_CTE,
  CTE,


  // -----------------------------
  // Updates
  // -----------------------------
  INSERT,
  BATCH_INSERT,
  DELETE_OPERATOR,
  UPDATE,


  // -----------------------------
  // Schema
  // -----------------------------
  CREATE_TABLE,
  CREATE_TABLE_AS,
  BATCH_CREATE_TABLE_AS,
  CREATE_INDEX,
  ALTER,
  CREATE_SEQUENCE,
  CREATE_VIEW,
  CREATE_SCHEMA,
  CREATE_MACRO,
  DROP,
  PRAGMA,
  TRANSACTION,
  CREATE_TYPE,
  ATTACH,
  DETACH,


  // -----------------------------
  // Helpers
  // -----------------------------
  EXPLAIN,
  EXPLAIN_ANALYZE,
  EMPTY_RESULT,
  EXECUTE,
  PREPARE,
  VACUUM,
  EXPORT,
  SET,
  LOAD,
  INOUT_FUNCTION,
  RESULT_COLLECTOR,
  RESET,
  EXTENSION
};

让我们看一个简单inner join的例子：物理执行计划最上头是投影算子PROJECTION，然后其左子树是HASH_JOIN算子，HASH_JOIN两个子算子分别为两个顺序扫描SEQ_SCAN：

基于物理执行计划构建出pipeline，真正执行的是pipeline。

2、物理执行计划如何构建pipeline？

2.1什么是MetaPipeline

MetaPipeline 表示一组都具有相同Sink的Pipeline。Source为输入，Sink为输出，Other Node就是其他节点，将一个物理执行计划树转换成多个pipeline。一个pipeline包含一个source和一个sink以及若干个operators。

pipeline还存在一定的依赖关系，hashjoin节点必须依赖build端的pipeline产生的数据才行，所以就需要MetaPipeline构建多个pipeline依赖关系，最后执行时仅关注pipeline就可以。

以1中的例子介绍pipeline的构建过程：

2.2 Pipeline的构建

1）最开始由Executor：：InitializeInteral函数创建一个MetaPipeline。该MetaPipeline的sink为NULL，vector<>pipelines容器创建一个pipeline，该pipeline的sink为NULL。

2）接着调用root_pipeline->Build(*physical_plan)使用上面的MetaPipeline继续构建pipeline

3）physical_plan为RESULT_COLLECTOR，Build会调用对应operator的Buildipelines,即调用PhysicalResultCollector::BuildPipelines，PhysicalResultCollector为PhysicalOperator的子类。

将当前operator即PhysicalResultCollector作为当前pipeline的source，如上图所示。

4）接着在调用CreateChildMetaPipeline创建一个child_meta_pipeline，sink节点为当前节点，即PhysicalResultCollector：并构建出和上一个pipeline的父子关系

代码：

5）紧接着使用child_meta_pipeline继续构建pipeline。下一个算子是PROJECTION：PhysicalProjection，它没有重写基类的BuildPipelines，那么就调用PhysicalOperator的BuildPipelines:

projection不是sink，并且它的子节点不为空，所以在当前pipeline添加一个算子即PhysicalProjection：也就是将PhysicalProjection放到当前pipeline的operators容器中

6）children[0]->BuildPipelines构建当前算子PhysicalProjection子节点的pipeline。此时到了HashJoin，即需要调用PhysicalHashJoin的PhysicalJoin::BuildJoinPipelines继续构建pipeline

首先将HashJoin添加到当前pipeline的operator容器中（因为作为探测端的pipeline）;然后保留一份当前MetaPipeline中的所有pipeline到pipelines_so_for后面使用；接着构建build端的MetaPipeline：CreateChildMetaPipeline函数完成：主要是构建一个pipeline，sink为当前PhysicalHashJoin,source为PhysicalTableScan：此时构建的pipeline如下图所示：

7）然后调用op.children[0].BuildPipelines继续build探测端的pipeline，实际上将左表的PhysicalTableScan设置到探测端pipeline的Source中。如上图所示。

8）外连接需要使用步骤6）保存的pipeline，构建一个childpipeline：

即使用Metapipeline2的pipeline再构建一个childpipeline，需要将PhysicalProjection操作符算子也加进去，此时结构如下图所示：

9）接着会添加依赖，都是在CreateChildPipeline函数中完成。对于当前的metapipeline，即MetaPipeline2它有两个pipeline：pipeline[0]：probe端；pipeline[1]：child pipeline。首先将当前pipeline（pipeline[0]）放到dependencies[child_pipeline]中；然后调用AddDependenciesFrom(child_pipeline, last_pipeline, false)继续添加依赖关系，从last_pipeline开始继续向dependencies中添加。

例如，当前metapipeline中有n个pipeline，下面pipeline[1]为起使pipeline，pipeline[m]为dependant，那么会将中间所有的pipeline都添加到dependant依赖数组里面。

pipelines[0] 
....
pipelines[s]   ---> start
.....
pipelines[m]   ---> dependant


pipelines[n-1]

结构：unordered_map<Pipeline *, vector<Pipeline *>> dependencies;完成依赖后：

pipelines[m] : [pipelines[s]......pipelines[m-1]]

由于这里的s=0;m=1所以依赖关系为：pipelines[1] : [ pipelines[0] ]，其中pipeline[1]就是child_pipeline。如此：child_pipeline : [probe pipeline]，表示probe pipeline依赖child_pipeline.

10）返回到1），此时进入root_pipeline->Ready()

以8）的metapipeline2中的pipeline[0]为例，反转前：

反转后：

11）总结：8）中为所有Metapipeline和pipeline：

第一个Metapipeline:

{pipelines[1], children[1]}

第二个Metapipeline:Children Metapipeline:

{pipelines[2], children[1]}

第三个Metapipeline:children metapipeline:

{pipelines[1], children[0]}

注意：表示的是数组大小

12）最后再次回到1）Executor::InitializeInternal函数，会从root_pipeline(他是metapipeline)，递归调用所有的metapipeline的pipelines数组，将pipeline汇总到root_pipelines中：

root_pipeline->GetPipelines(root_pipelines, false);
//vector<shared_ptr<Pipeline>> root_pipelines;

这就是pipeline的一个生成过程，下期介绍这些pipeline是如何调度的