MonetDB/X100: Hyper-Pipelining Query Execution 论文解读

这是关于MonetDB执行引擎的一篇paper，总体分为了2个部分，第一部分主要讲了下modern cpu的工作原理并给出了一个TPC-H Q1的例子，从这部分中我们便可以清晰的看到为什么向量化的执行方式如此有意义。第二部分则主要介绍了MonetDB X/100这个新的向量化执行引擎。

这篇paper被引用的极为广泛，启发了后续很多列存数据库在执行引擎上的设计思路。个人感觉第一部分尤其有意义，如果想入门下列存/向量化执行，看看第一部分应该就有些概念了。

介绍

由于现代CPU具有了并行流水线能力，如果能够找到相互独立的并行任务，就可以充分利用CPU的高速计算能力，对于像OLAP这种workload，应该会有针对大量（独立的）数据进行密集计算的需求，因此潜在的可以更好的利用CPU，追求更高的IPC (Instructions per cycle)。

但是研究显示当时的数据库系统的IPC普遍是比较低的，因此paper基于TPC-H这种决策支持分析query进行了profiling，发现其原因是大部分DBMS的执行架构使得编译器无法进行充分优化来利用CPU的并行能力(super-scalar)，尤其是Volcano iterator这种tuple-at-a-time的模型增加了很多的interpret成本，由于每次只处理一行，也无法实现基于独立数据的CPU并行。

其实volcano模型在Graefe提出的那个年代是很先进的，当时的硬件架构还处于cpu计算能力较弱 + cache很小，主要的处理瓶颈在disk IO上，volcano的这些额外cpu代价相对来说就不那么重要。

paper然后分析了MonetDB/MIL原有的column-at-a-time的执行方式，这种模型需要在各种计算之间对全列数据做物化，虽然避免了interpret的成本，但也在执行中引入了大量的memory IO，profiling显示执行速度严重受限于memory IO bandwidth，从而影响了CPU的执行效率。

因此MonetDB/X100这种向量化模型实际是tuple-at-a-time的volcano和column-at-a-time的全物化之间的折中方案，很好的提升了CPU执行效率。

现代CPU

CPU的并行能力

上图给出了10年间CPU执行速度的提升，可以看到当时还是符合摩尔定律的。同时CPU引入了pipelining的能力：

1993 Pentium 5-stage pipeline, picture from wikibook

从上图可以看到，1条指令的执行划分为了5个stage，就像流水线上的工人一样，每个stage的工人处理完一条instruction后立即开始对下一条instruction的同一个stage进行处理，这样就让CPU的各个组件随时都在工作，提升了CPU的时钟频率。

这种pipeline也带来了2个问题

1. 如果一条instruction的执行必须依赖于前一条Instruction的结果，则它无法立即跟着前一条进入流水线，只能等待前一条走完整个pipeline。
   
2. 对于IF a THEN b ELSE c 这样的分支语句，CPU需要去预测条件的true/false，例如它预测a为false把c立刻跟随a送入流水线，但执行一些stage之后发现a是true（branch mispredicted)，但c以及后续指令已经进入了，只能flush掉整个pipeline，再从b重新开始。很明显pipeline越长，这个成本越高。而反应到数据库系统中，每个tuple的数据各有不同，这个条件a的值是很难提前预测的，这会影响到query的执行性能。
   

后来又出现了super-scala的CPU，也就是有着多条pipeline，这也使得CPU可以把多个各自独立的job同时并行执行。有了super-scala的处理器后，IPC就可以 > 1了。

现代cpu形态各有不同，有些(Itanium2)有着很多pipeline但每条的stages比较少，有些(Pentium4)则pipeline较少，但每条pipeline的stages很多，无论是哪种，CPU在任意时间的理论最大吞吐能力就是： num(pipeline) * num(stage)，当然这只是理想情况。

当然，大多数编程语言（包括C/C++)都不需要编程人员去指定指令之间的独立性，这需要由compiler来完成，因此compiler的优化能力对于cpu的充分利用就变得很重要，其中一个很重要的优化就是loop pipelining（这也是vectorize所追求的)：

对一个数组A中的每个元素a（元素间独立），由2个相关的计算任务F() -> G() ，且F()的执行需要2个cpu cycle，则loop pipelining可以将

F(A[0]),G(A[0]), F(A[1]),G(A[1]),.. F(A[n]),G(A[n])

转换为

F(A[0]),F(A[1]),F(A[2]), G(A[0]),G(A[1]),G(A[2]), F(A[3]) ...

这样当G(A[0])开始执行时，F(A[0])的结果刚好已经得到了！

这样的优化也是为何现在很多的数据库系统即使没有使用SIMD，也要先实现基于列存的向量化，就是为了利用loop pipelining的这种能力。

进入到数据库系统看下这条SQL

SELECT oid FROM table WHERE col < X;
控制X使得predicate的选择率在[0 - 100]间分布

可以看到，手工重写后的版本将不带有分支的if语句，因此其执行效率更高，且和选择率无关。(Itanium2自带编译优化，可以将branch version -> predicated version)

CPU cache

cpu所执行的指令中，大约30%是memory load/store，用来访问物理内存中的数据，这时就是我们常说的cache miss，这会产生至少50ns的物理延迟，而对于3.6GHz的高频CPU来说，这就是180个cpu cycle，因此只有当要访问的大部分数据都在cache中时，cpu才有可能获得最大的计算吞吐。

数据库很多算法是有严重的cache miss问题的(iterator/hash table probe...)，因此才出现了很多cache-conscious的数据结构(cache-aligned B-tree)或算法(radix-partitioned hash join...)

总的来说，现代cpu已经变得非常复杂，不同的cache命中情况，不同的分支预测情况，能够独立运行的job情况等，会使得cpu的执行效率产生巨大（几个量级）的差异。

Micro benchmark

本节针对TPC-H(1S) Q1的执行进行了分析，选择Q1是因为只想关注最基本的表达式计算，不考虑像join这样更复杂的关系运算，也不受太多查询优化的影响。

注：由于paper所在的年代较早，硬件特性与指标方面与现在已经发生了一些变化，但总体思路还是有参考意义的。

SELECT l_returnflag, l_linestatus,
       sum(l_quantity) AS sum_qty,
       sum(l_extendedprice) AS sum_base_price,
       sum(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
       sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
       avg(l_quantity) AS avg_qty,
       avg(l_extendedprice) AS avg_price,
       avg(l_discount) AS avg_disc,
count(*) AS count_order
FROM lineitem
WHERE l_shipdate <= date '1998-09-02'
GROUP BY l_returnflag, l_linestatus;

Q1的特点是

1. 选择条件的过滤性很差，对于6M的lineitem行，剩余还有5.9M
   
2. 分组很少（4组），因此可以使用hash aggregation在cpu cache中高效完成运算(hash table很小，不需要额外的memory IO）。
   

Query 1 on MySQL

对于常规的解析执行方式，由于要做到通用，需要考虑各种可能的数据类型和不同的tuple形态，因此无论是从tuple中解析出某一个字段，还是对该字段进行类型相关操作，都有很多额外成本，当tuple非常多时，这种成本被进一步放大，导致实际执行的有效操作非常少：

上图是MySQL4.1中对于Query 1的执行分析，cum是该操作累计执行时间，excl是该函数自身执行部分所在比重（排除它调用的其他函数），calls是调用次数，后两列是每次call执行的平均指令数和IPC。

可以观察到，上图中5个加重的函数是实际执行运算的部分，它只占总体执行时间的10%，还有28%用在了hash table lookup中，另外62%则被rec_get_nth_field这类的辅助解析函数等占用。

在MIPS R12000 CPU下，仔细查看Item_func_plus::val这样的实际计算函数，每次加法使用了38个instructions，而这种CPU每个cycle可以执行3个普通的整型/浮点型addition指令和一个load/store指令，每个操作的延迟大概在5cycle。一个简单的加法运算在RISC指令集上：

+(double src1, double src2) : double
=>
LOAD src1,reg1
LOAD src2,reg2
ADD reg1,reg2,reg3
STOR dst,reg3

这套操作中比较重的是LOAD/STORE指令，因此一个加法运算大概是3个周期（3个LOAD/STORE），而在MySQL中得到的，则是 #ins/IPC = 38/0.8 = 49，要这么多个cycle才能完成一个加法运算！

其中一个原因就是MySQL的这种tuple-at-a-time的方式无法实现loop pipelining的优化，每个指令都要等待其完成(5 cycle)才能执行下一个，这样光是这4个指令就要4*5 = 20个cycle了。剩下的29个cycle则来源于函数的调用，stack的建立，入栈出栈这些。

paper中同样测试了一款流行的商业数据库，得到了类似的结果（高度怀疑是Oracle）。

Query 1 on MonetDB/MIL

MonetDB的数据存储方式是BAT(Binary Association Table)，每列数据都以二元组[oid,value]来表示，oid表示tuple的唯一标识。

不同于关系运算，MIL的代数运算方式是固化的，每个运算有固定的输入格式和输出，复杂的运算可以由简单运算组合而成，例如

extprice * (1 - tax)
=>
tmp1 := [-](1,tax)
tmp2 := [*](extprice,tmp1)
[*]()/[-]()称为multiplex operator，把function映射到对应的BAT上
计算的输入/输入都是BAT的形式，且都是物化的

对MonetDB/MIL执行query 1，可以看到一共20个MIL的操作占用了99%的执行时间（MySQL只有10%)。仔细观察下第2列和第4列，会发现当数据量较大时(SF = 1)，执行受到了memory bandwidth的限制，最大只能到500M/s，而如果数据量小(SF = 0.001)，所有数据到保存在cache中，cpu <-> cache间的吞吐可以达到1500M/s。对于一列计算，输入+输出一共占用16 + 8 = 24个字节，因此500MB的带宽意味着每秒20M的操作次数，如果是1533MHz CPU，则等于每个操作占用75 cycles，还不如MySQL的49 cycles！

因此这种column-at-a-time的执行方式产生了正反两面的影响：

1. 正面上，不再有tuple-at-a-time的解析成本，由于操作的是整个column vector，可以充分利用loop pipelining
   
2. 负面上，由于运算结果的全量物化，导致在计算中会经常需要大量的memory IO操作。
   

Query 1 on hand-write code

为了得到在现代CPU上最为理想的执行性能作为baseline，paper中实现了一个c代码的UDF：

static void tpch_query1(int n, int hi_date,
                        unsigned char*__restrict__ p_returnflag,
                        unsigned char*__restrict__ p_linestatus,
                        double*__restrict__ p_quantity,
                        double*__restrict__ p_extendedprice,
                        double*__restrict__ p_discount,
                        double*__restrict__ p_tax,
                        int*__restrict__ p_shipdate,
                        aggr_t1*__restrict__ hashtab)
{
  for(int i=0; i<n; i++) {
    if (p_shipdate[i] <= hi_date) {
      aggr_t1 *entry = hashtab + (p_returnflag[i]<<8) + p_linestatus[i];
      double discount = p_discount[i];
      double extprice = p_extendedprice[i];
      entry->count++;
      entry->sum_qty += p_quantity[i];
      entry->sum_disc += discount;
      entry->sum_base_price += extprice;
      entry->sum_disc_price += (extprice *= (1-discount));
      entry->sum_charge += extprice*(1-p_tax[i]);
    }
  }
}

在MIL中，目标列以BAT[void,T]的形式存储，其中void(virtual-oid)是不实际存储的，因此输入的就是普通array。通过__restrict__ 描述数组间不重叠，这样就可以应用loop pipelining，从本节第一张图可以看到，手写的执行时间只有0.22秒，远优于MySQL和MonetDB/MIL。

而从图中也可以看到，MonetDB/X100也可以达到非常好的性能，只有baseline的2倍执行时间，下面看下X100的介绍。

X100 : A Vectorized Query Processor

X100是一种全新的执行引擎，其思路是结合Volcano这种tuple-at-a-time的流水线和MIL的column-at-a-time的无流水线全量物化，做一个折中，每次处理一个批次的列数据。

为了提高性能，它考虑了整体体系架构的各个可能产生bottleneck的环节：

• Disk I/O
  

通过底层列存，减少不必要的数据存取，降低带宽要求，且可做基于列的轻量级压缩

• Memory
  

在内存中也保持垂直分片（列存）和压缩的状态，与磁盘上相同，减少内存的占用。

• Cache
  

利用Volcano style的vectorized执行，vector是一个足够小的基本处理单元，可以fit in cache中，减少与memory的交互，提高效率。只有当数据进到cache层才进行解压缩/压缩，这时由于cpu和cache之间的高带宽，这种运算效率很高。X100的设计原则是尽量只在cache chunk内做random数据访问。

• CPU
  

由于每个vector是针对一列的切分chunk，vectorized primitives符合loop-pipeling的优化条件，可以重复利用CPU的并行流水线。而且可以针对复杂表达式，通过对vectorized primitives做组合，进一步提高执行效率。

上图中是当时X100的架构，由于ColumnBM还没有开发完，底层存储仍然是BAT的形态，但执行层则使用了X100的关系运算，包括一系列的向量化原语，它们针对in-memory的BAT执行计算。

Aggr(
  Project(
    Select(
      Table(lineitem),
      < (shipdate, date('1998-09-03'))),
    [ discountprice = *( -( flt('1.0'), discount),
extendedprice) ]),
[ returnflag ],
[ sum_disc_price = sum(discountprice) ])

执行器采用了volcano的pipelining方式，处理粒度是一个vector(1000 values)。

Scan操作从BAT中一次获取一个vector进入Select算子，计算的结果会生成一个selection vector来描述其中哪些行被选中了，这里原始的vector不会被in-place修改，而是在后续的计算中，利用map-primitive来完成只针对筛选行执行计算，计算结果也到新vector中对应的原始位置，后续计算也会继续利用selection-vector判断新vector中有效数据在哪里，这就要求selection-vector要一直向上传递。只所以选择这个方案是因为保持输入vector不变作为输出vector，比重新拼接出新的vector要更加高效。(SQL Server的column index也采用了这个方案）

Q1的执行流程如下图所示：

上图中map_sub_flt_val_flt_col就是一种基本计算的原语，实现两组flt类型vector的sub操作，实现中需要针对各种类型各种组合，实现很多这种原语，这些向量化原语在MonetDB中是通过模板的方式，自动生成对应各种类型的宏定义。

X100 Algebra

上图列出了X100支持的各种运算，其中Table表示物化的数据，而Dataflow则表示pipelining的流式结果。

Order/TopN/Select算子不会改变输入dataflow的shape，其他算子则会生成新shape的vector。这里各种算子的实现方式其实不太重要，也就不细说了，paper中也就大概提了下，有兴趣的同学可以自己翻阅下paper或其他资料。

Vectorized Primitives

这种向量化的计算原语与解析执行正好相反，其优势是低的灵活性，每个执行原语只针对特定类型的定长列数组做计算，不关心对tuple的解析或者列的类型（已指定），这种简单的形式可以让compiler做很激进的loop pipelining优化，例如：

map_plus_double_col_double_col(int n,
double*__restrict__ res,
double*__restrict__ col1, 
double*__restrict__ col2,
int*__restrict__ sel)
 {
  if (sel) {
    for(int j=0;j<n; j++) {
      int i = sel[j];
      res[i] = col1[i] + col2[i];
    }
  } else {
    for(int i=0;i<n; i++)
      res[i] = col1[i] + col2[i];
  }
}

这个原语只针对2个double类型的列数组指向加法操作，在上下两个代码分支中，都可以做到紧凑的循环计算，sel参数就是selection_vector，所有X100的向量化原语都有这个参数。

X100中实现了几百个这样的primitives，此外还支持组合原语的方式，例如：

/(square(-(double*, double*)), double*)

这种组合原语会被单功能的简单原语更加高效（paper中是2倍），原因是更高的指令组合，如果是一个计算，可能要执行load -> exec1 -> store这样的操作，但如果将计算组合起来，可以将第一个计算的中间结果保存到寄存器中，从而实现load -> exec1 -> reg -> exec2 -> store这样的流程，这样load/store操作的成本就被均摊到了多个运算中。

这里vector size是一个需要权衡的点，过小则无法充分利用loop pipeling，且volcano的流水线中，函数调用等的成本相对升高，过大则无法fit in cache中，效率也会降低。

Data Storage

初始的实现是基于BAT做存储，每个BAT(列）是一个文件，但这种更新一个tuple的成本就很高，会有多个文件的随机IO。改进的方式就是使用main + delta的经典方案(C-store/Kudu)，main仍是按列存（支持字典压缩）+ 不可修改。delta保存在内存中，更新append进来，周期性和main做compaction。

delete通过把tupleid加入到del列表中完成，而insert则把数据加入到内存的各个delta columns中，ColumnBM的存储引擎会把各个delta columns存储在一个chunk中(PAX格式），update = delete + insert。

对main部分的每列，ColumnBM在存储时会将其切分为多个chunk，对每个chunk建立简单的min/max的系数索引，帮助range predicate的快速过滤。而delta是在内存中，且数据量小，不需要索引。

总结

MonetDB应该算是向量化执行引擎的始祖？这篇paper用非常严谨的分析说明了传统的volcano那种tuple-at-a-time的执行方式在分析型场景下是如何的低效，或者纯粹的按列物化的方案所存在的memory bound问题，因此采用了折中方案：

1. 流水线的执行算子
   
2. 小的，连续内存的，常驻cache的，固定类型数组作为计算单元(vector)
   

各取所长，便是vectorized执行引擎了，具体的实现要依赖大量vectorized primitives，在其中也可以利用SIMD做进一步优化。