一、MatrixOne整体架构
MatrixOne早期的架构是一个典型的share nothing架构，数据存放在一个Multi Raft集群上面，数据的每一个切片存在一个Raft上面，不同的Raft Group之间的数据是完全没有重叠的。
早期架构存在着一些无法解决的问题，比如在扩展性上，每扩展一个节点，就需要同时扩展存算的资源，因为计算和存储没有完全分开。而且每扩展一个节点，需要大量的数据迁移工作。另外因为每一份数据都要保存至少 3个副本，从扩展节点到完成的时间会非常久。在性能方面，Raft协议所包含的leader角色，容易造成热点；在性能较差的存储下，数据库整体性能下降会超过预期；多种引擎各自用途不同，性能各异，无法有效应对HTAP场景。成本方面，数据保存3副本，随节点规模增长，成本不断攀升，云上版本更甚；只有高配存储才能发挥数据库的预期性能。
为了更好地满足客户的需求，我们升级了新的架构。
升级后的架构从整体上看，分为三个部分：
- 最上面是计算层，计算层里面的每一个单位是一个Compute Node (CN) 计算节点。
- 在计算层下面是数据层，由Data Node (DN) 组成。
- 再下面是File service，支持各种不同的文件系统。
下面分别详细介绍一下每一部分。
- 先从最底下的File service说起。File service支持各种不同的文件系统，比如用户自己的本地的磁盘，NFS、HDFS、或者对象存储。File service对上层提供一个统一的接口，对用户来说，他并不需要关心最底下的数据本身是存在什么样的介质上面。
- 这边还会有个log service，因为我们知道在file service插入数据的时候，只能一整块一整块地写，尤其像S3，一个object非常大，不可能一行一行地写，一般是积累到一定的量再往里面写一个整块，那么这些还没形成整块的部分，会放在一个单独的log service里面。log service是一个Multi Raft的集群。已经形成整块的部分的数据完整性和一致性是用S3或者HDFS的功能来保证。还没有形成单个block的数据的一致性完整性则是通过Multi Raft group来保证。
- 再上面是存储节点DN，存储节点里只存放的是元数据信息，比如每个表会分成大的segment，每个segment又会分成很多小的block。比如用S3来存放数据，元数据里会存放每个block，存放对应的具体S3 object。还有其他元数据信息比如row map，或者是次级索引bloom filter这些信息也会存在DN上。这里可以看到有多个DN，它们是怎么分布的呢？比如一张表有一个主键，那么我们可以根据主键来做一个分区，目前是按hash来做分区，也可以按range来做都是没有问题的。对于数据规模较小的客户来说，存放元数据其实只需要一个DN就足够了。
- 再上面就是计算节点CN，具体执行计算任务的节点。CN可以分成各种不同类型，比如专门做TP查询用的，专门做AP查询用的，专门做streaming用的，还有一些对用户不可见的数据库自己后台的一些任务。
- 还有一个组件叫HA Keeper，跟ZooKeeper功能类似，在节点之间互相通知上线下线这些信息，维护整个集群的可用信息。
新的share storage的架构优势在于：
- DN节点不保存任何数据，只保存元数据信息。这样就不会让单个DN成为瓶颈，如果需要做弹性扩缩容，比如DN要新增或者删除一个节点。做数据迁移只需要交换一些元数据的信息，不需要把所有数据都做搬迁的操作，这大大简化了数据迁移量，并提高了扩缩容的效率。
- 完整性和一致性主要是通过S3来保证，S3本身就具备这一功能，并且从成本上来看也是非常低的，比用户自己去搭建Multi Raft集群的成本会低很多。
- 计算任务也可以很好地分离开，比如TP和AP的计算，可以放在不同的CN节点上去做。因为现在数据已经不跟DN绑定在一起，所以计算的节点也可以完全解耦。
我们来看一个典型的查询里面读数据的操作是怎样的。数据查询从一个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：把一个SQL语句解析成一个AST树。
- 第二个是planner：把AST树转化成一个逻辑计划。
- 第三个是optimizer：把逻辑计划通过各种优化器规则或者是通过一些基于代价估算的方式转化成更好的逻辑计划。
- 最后是execution：把具体的逻辑计划转化成可执行的pipeline，然后去具体的CPU上面执行。
语法解析器（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是如何设计的。
（1）第一部分是减少IO
包括以下部分：
- 列裁剪，读一张表，有很多列，但我实际上只用到其中一列，那么其它列是不用读取的。
- 谓词下推，就是把一些过滤条件直接推到读取数据这一部分，这样可以尽量少的读取数据。
- 谓词推断，主要会影响TPC-H里面的Q7和Q19，在后面会再举例详细介绍。
- Runtime filter ，比如大表跟小表做join操作，在小表构建完hash表之后，可能hash表的计数非常小，这样我们可以直接通过hash表里面不同的词去大表里面通过runtime或者元数据信息进行过滤，这样在运行时就大大减少了需要读取的大表的block数量。
（2）第二部分是减少计算
它并不能减少实际从磁盘里面读取的数据，但是会在计算过程中减少计算量，和减少中间结果的数据量：
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张表时，可以先用贪心的方法来做一个剪枝操作，让搜索空间大大减小，在贪心法之后再做动态规划。
贪心法分三步：
- 第一步是确定事实表和维度表，因为一般的OLAP查询的数据通常会把表分成事实表和维度表，之间用维度表的主键做join。因此拿到一个查询之后，我们可以把事实表和维度表给找出来。
- 下一步事实表先与其维度表join成子树，因为事实表维度表之间始终是通过维度表的主键去做join。做这样的join的结果的函数始终是不大于他本来输入的函数，所以做这么一个join并不会造成很多OLAP开销，不会造成数据膨胀。在做事实表维度表join的过程中，我们会考虑事实表先与过滤性好的维度表做join。就优化器而言，越早减少数据量是越好的。
- 最后一步是在子树之间，像TPC DS的表，会有多个维度表，维度表互相之间都不是以主键来做join。那么在子树之间，我们再使用经典的join order的算法，比如动态规划等等。
这样我们把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操作。
发送也分为两类：
- connector算子是一对一的发送。
- dispatch算子是一对多的发送。
dispatch会有很多不同的模式：
- 一种是广播的模式，假设S表是一个很小的表，构建完hash表之后，会把hash表广播到不同的pipeline出来去做计算；
- 一种是做shuffle，假如S表和R表都比较大，因此要做shuffle join，那么直接会通过一个shuffle dispatch算子把数据发送到不同的对应的一个pipeline上面去。
对于OLAP系统，从语义上来说通常跟SQL本身没什么关系。但是OLAP的分析性查询会是比较复杂的计算任务，有一些SQL能力是必须具备的，比如多表join、子查询、窗口函数，还有CTE 和Recursive CTE，以及用户自定义函数等。MatrixOne目前已经具备这些能力。
