MatrixOne：HTAP数据库中的OLAP设计

发布时间：2025-11-05 14:12:29 来源：技术快报 作者：IT科技

一、数设计MatrixOne整体架构

MatrixOne早期的据库架构是一个典型的share nothing架构，数据存放在一个Multi Raft集群上面，数设计数据的据库每一个切片存在一个Raft上面，不同的数设计Raft Group之间的数据是完全没有重叠的。

早期架构存在着一些无法解决的据库问题，比如在扩展性上，数设计每扩展一个节点，据库就需要同时扩展存算的数设计资源，因为计算和存储没有完全分开。据库而且每扩展一个节点，数设计需要大量的据库数据迁移工作。另外因为每一份数据都要保存至少 3个副本，数设计从扩展节点到完成的据库时间会非常久。在性能方面，数设计Raft协议所包含的leader角色，容易造成热点；在性能较差的存储下，数据库整体性能下降会超过预期；多种引擎各自用途不同，性能各异，无法有效应对HTAP场景。企商汇成本方面，数据保存3副本，随节点规模增长，成本不断攀升，云上版本更甚；只有高配存储才能发挥数据库的预期性能。

为了更好地满足客户的需求，我们升级了新的架构。

升级后的架构从整体上看，分为三个部分：

下面分别详细介绍一下每一部分。

新的share storage的架构优势在于：

我们来看一个典型的查询里面读数据的操作是怎样的。数据查询从一个CN节点开始，首先去访问DN的信息，读取元信息，判断某张表在哪些block，甚至会先通过过滤条件用row map来做过滤，剩下的是实际上需要去读的那些block信息，拿到之后直接去那个file service下面去读取数据。

在这里DN节点只提供了元数据信息，基本上不会成为性能瓶颈。因为更多的数据是CN节点去直接向File service读取的，CN节点上面还会维护一个metadata cache。假设这个数据新鲜度还没有过期，甚至可以不需要访问DN，直接去File service下面拉取数据。

最后是存储引擎，TAE，T和A分别是指TP和AP，T和A表示它既有事务的能力也可以很好的处理分析性查询，用同一套引擎同时支持AP和TP。

在结构上，我们实际上还是可以把它看成是基于列存，不同的列之间，数据仍然是单独分开存放的，每一列会按照8192行分成一个小的block。对于大多数的数字类型的block，8192行数据可以在L1 cache里面直接装下，在后面的批量计算的时候会对计算引擎比较友好。多个block会组成一个segment，segment的作用是假设这个表它有主键或者排序键的情况下，一个segment内部会通过排序键和主键去做排序，这样数据存储在每一个segment内部是保持有序的，但segment之间可能会有重叠。这跟LSM的存储有些相似的地方。但如果我们之后做了partition功能，可能会把一个partition内的所有segment也去做一次compaction操作。即把它们重新拿出来，做一个归变排序再放进去。

以上就是MatrixOne最新的架构。

二、MatrixOne OLAP引擎

MatrixOne的计算引擎分为四个部分：

语法解析器（Parser）

各大开源数据库大多都不会去手写一个parser,至少是用MySQL或者PG的parser。比如DuckDB就是直接使用Postgres的parser代码。即使我们不直照搬，也可以用一些YACC的工具去生成一个parser。测试之后发现用YACC生成的parser，并不会成为性能瓶颈，它的耗时非常少，所以我们没有必要去手写一个parser。（除了ClickHouse的parser是手写的）。parser生成AST树之后，会通过逻辑计划器，把AST树转换成一个可以执行的逻辑计划。

逻辑计划器（Planner）

逻辑计划器主要包含两个部分：Bind(Algebrizer) 和子查询消除。因为我们并不支持像SQL Server一样将子查询转换成apply join，或者像MySQL一样完全从父查询里面拿出一行，再带入子查询里面，把子查询完整地执行一次。考虑到在AP查询的场景，这样的一个执行计划是不可接受的，就干脆完全不支持这种apply join的方式，所以我们在planner这一步，把子查询的消除给做掉。

优化器（Optimizer）

优化器部分，通常会有一个RBO基于规则的优化，基于规则对大部分查询已经是够用的。因为优化器通常分为两种，一种是减少数据IO，它会减少实际从磁盘文件系统读取的数据量；还有一种是对于CPU，在计算的过程中减少计算的代价。下面会具体举一些例子来说明MatrixOne是如何设计的。

包括以下部分：

它并不能减少实际从磁盘里面读取的数据，但是会在计算过程中减少计算量，和减少中间结果的数据量：

join order的join定序。通常使用TPC-H做OLAP 的benchmark，join order会影响很多不同的查询，如果join order做的不好，这些查询都可能会以数量级的变慢。

聚合函数下推和上拉的操作。假设聚合函数是在一个join上面，如果是先做join，之后再去做聚合，那么在join这里，数据可能会膨胀的非常多。但是如果可以把聚合函数推到join下面去做，即在join之前先聚合，数据已经减少很多，这也可以大大的减少计算。

简单介绍谓词推断：

谓词下推是已经确定显式的可以下推的一个位置。但谓词推断可能是需要做一些逻辑上的变化之后，才能得到一些新的谓词，这些新的谓词才可以下推下去。比如TPC-H Q19的过滤条件是三个很长的谓词用or连接起来，通过观察实际上这三个or里面有很多共同的部分，可以把共同的部分提取出来，变成右图的样子。可以观察到，首先part这张表上面有一个可以下推的谓词，lineitem这张表上面也有两个可以下推的谓词。这样可以先把这两个谓词下推到每个表的table scan上面去。然后还多出来一个在part和lineitem上面用主键做连接的谓词。如果原来不对这个执行计划做优化直接去执行，可能会先做一个笛卡尔积，再去做过滤操作，这样的效率会非常低。现在我们可以把它变成一个join操作。

简单介绍一下MatrixOne使用的join order算法：

在各大开源数据库上，join order的算法实现主要包括贪心法和动态规划。其中动态规划有很多不同方法，也有很多论文可以参考。但是动态规划存在一个问题，当表的数量稍微多一些，状态搜索空间就会以指数级膨胀。比如StarRocks的文档里面提到过，10个表以上的计算就没办法使用动态规划来计算，只能使用贪心法来计算。

我们在MatrixOne里对这个问题做了思考，在大于10张表时，可以先用贪心的方法来做一个剪枝操作，让搜索空间大大减小，在贪心法之后再做动态规划。

贪心法分三步：

这样我们把Join order的算法从之前只能10表以下做动态规划，扩展到10个事实表之下都可以使用动态规划来做。

举个例子，TPC-H Q5中有customer、 orders、lineitem、supplier、nation、region这么六张表，它是一个比较典型的星型结构。

我们将orders和region这两个表标红，因为它们上面有过滤条件，需要单独考虑。然后对于本身的join条件，标上一些带箭头的线。这里从事实表到维度表，用维度表的主键做join的条件。可以看出，一共有5个条件都是用主键来做join，其中还有比较特殊的一个条件，用画的虚线标记，它的两边互相都不是用主键来做的join。

这个join算法下一步还会有一个联合优化，最后可以跟谓词推断联合使用，形成一个新的优化。因为可以看到supplier和customer有一个join条件，supplier跟nation也有一个join条件，用的都是同一个类型T来做决定。我们可以推导出来nation和customer表之间是以类型key做为join的条件，因此我们用黄色的选项表示。

最后实际生成的最优的join顺序是从region开始，先跟nation表join，再跟customer表join，再跟orders 表join，再和lineitem表 join，最后跟supplier表 join。这是我们新推断出来的条件。那么为什么是走这么一个路线，而不是先把上面这一条路join完成后再join下面这边的表，是因为我们考虑到orders，region这两张表都有过滤条件，放在一起过滤效果会更好。这样会让lineitem这张表的行数减少的速率更快。

执行器（Execution）

最后是执行器部分，从逻辑计划转化到实际可执行的pipeline，执行器的好坏对OLAP系统的执行效率影响是非常大的，下面会做详细介绍。

众所周知，执行器有一个经典的火山模型。对于每一行它是一个典型的pull模型，从最上层的计算的operator开始，每次调用一个NEXT函数从下面的节点去拿一行新的数据出来，做完计算之后，再等待更上层的那个计算节点去调用next从它这里取走。

火山模型存在一些问题，首先它并没有做并行化，而是一行一行地处理；而且每一行在不同层级之间做一次调用，实际上会产生虚函数的开销。因为next在不同编程语言肯定是要做函数重载，就算是虚函数开销也很有可能是比实际计算的开销还大，在整体开销上会占相当的比例；同时对缓存也是非常不友好的，因为一行数据会跑多个不同的operation，可能在取下一行的时候，原来的缓存已经被清洗掉了。

MatrixOne的执行器是基于push模型，可以把几个连续的operator组成一个流水线，而且流水线里面流动的数据，并不是一行一行的数据，而是前面提到的TAE存储引擎里面的一个block，包含8192行数据，对于一般的数字类型是可以直接放进L1 Cache里面的，对缓存非常友好。每一个operator每一次要处理完这8192行才会喂给下一个operator，再加上调度是从最下面的，实际读取每一张表的table scan 那个节点开始，往上面push。

对于push模型，是以数据为中心而不是以operator为中心，它的生成过程是对上一步的planner和optimizer生成的逻辑计划，作一个后续遍历，后续遍历之后就可以得到一个基础的pipeline结构，这个基础的pipeline结构还没有带上比如每台机器有多少个CPU或者需要在多少个CN节点上去执行这样的信息。在后面实际执行的时候，再动态地根据这些基础信息去做扩展。

举个简单的例子，假设一个简单的查询，有R、S、T三张表做join，其中R表是最大的一个表，S表和T表相对比较小，并且每个表都有过滤条件。对于一个典型的hash join，我们会把S和T这两张表去构建hash表，然后R表在这两个hash表上面依次去做探测（Probe）操作，得到join之后的数据。这么一个逻辑计划至少需要插上三个pipeline，S表的读取数据，做完过滤之后再建完hash表就在这里终止，T表也是在建完hash表之后就在join算子上面终结掉。但是最大的R表它始终是要做probe的，这张表的pipeline就可以往上走很多步。比如先做完过滤，再跟S表join之后，仍然以批量数量。然后这个批会继续往下走，在下一个join中，仍然在同一个pipeline的下一个operator里面，再跟T表做一个join。所以一个3表join通常会拆成3个pipeline出来。

右图还包括了数据并行的信息，比如S表可能会使用go语言里的三个协程并行的去读数据，再做合并操作，合并完之后构建hash表，T表也是用三个协程去并行的读，读完之后，然后送到这里的构建hash表。R表因为比较大，pipeline会展开出更多的实际pipeline出来。我们可以看到就是R表这个pipeline是不会被阻断的，通过hash operator之后，会继续进到下一个join节点。

如上图，我们的pipeline提供了这些算子，比较典型的有聚合、分组和各种join操作。这里把merge和connector、dispatch的颜色标识成不一样，因为它们和其它operator的区别是其它算子都是只能在一个pipeline的中间，接受的数据是从上一个算子传过来，发送的数据就直接发送给下一个算子去做后续的计算。而标成灰色的这一部分是在一个pipeline的数据的source或者sink，即入口或者出口的地方，比如merge它会去其他的pipeline去接收数据，把所有pipeline的数据合并成一个，返回给用户。同理group by或者order by算子，也会执行merge操作。

发送也分为两类：

dispatch会有很多不同的模式：

对于OLAP系统，从语义上来说通常跟SQL本身没什么关系。但是OLAP的分析性查询会是比较复杂的计算任务，有一些SQL能力是必须具备的，比如多表join、子查询、窗口函数，还有CTE 和Recursive CTE，以及用户自定义函数等。MatrixOne目前已经具备这些能力。

相关文章