The MemSQL Query Optimizer: A modern optimizer for real-time analytics in a distributed database

在前几篇文章中，我们已经介绍了数据库查询优化器最为经典和常见的一种框架：bottom-up DP-based search。

这种框架以System-R为代表，但其中DP-based search主要是针对join ordering这个优化的子问题的，由于无法做到像Cascades那样将query transformation融入其中，因此后继的很多开源/商业数据库系统在此基础上，采用2-pass的方式，即先做query rewrite(transformation)，再做cost-based search，例如IBM DB2，PostgreSQL，Oracle等。

这种2-pass的方法有个很明显的问题，就是transformation是基于heuristic的，也就是无法确保查询的变形一定会更优，导致要么过度变形，要么只能做一些简单且基本一定更优的rewrite（例如projection消除，join消除，filter下推...)，无法适当的扩展后续的搜索空间。针对这个问题，2-pass优化框架也在演进，今天我们要介绍的MemSQL就采用这样一种新的形态（Oracle也变为了这种方式 ）：即在做transformation时，要基于cost确定其是否可应用。

当然，本篇paper不止讲解了CBQT，还包括一些MemSQL优化器其他方面的介绍，包括一个有意思的heurstic based bushy join的方案。

MemSQL 介绍

MemSQL（现在改名叫SingleStore），是一个基于memory优化的，分布式SQL数据库，支持HTAP，目标是可以高并发的处理事务，并同时完成针对大量数据的复杂的实时分析查询。

• 数据有两种format : in-memory rowstore / on-disk columnstore，可以选择持久化的级别，例如fully durable、或允许丢失多大的数据量。
  
• 基于MVCC做并发控制，无读锁 + 轻量级写锁在，内存中使用lock-free skip list来做高并发的read/write。
  
• MemSQL的架构类似一个MPP(SQL Server PDW)，有两种role : aggregator node + leaf node ，table data按分片存储在leaf node中，包括hash/broadcast两种分布
  
• aggregator node是client入口，做parse+optimize+planning， 完成优化后将计划拆分成若干subplan分发到各个leaf node，leaf node负责数据存储和subplan执行。
  
• 分布式的执行计划DQEP，由一系列的step组成，step用SQL描述，使用ResultTables + RemoteTables来描述数据流转和存储中间结果。
  
• 在leaf node上，plan可以做codegen，编译为machinecode并cache下来，针对不同参数可以复用plan。
  

MemSQL optimizer

按paper中的介绍，由于MemSQL这种分布式内存数据库，需要应对TP/AP的不同挑战，因此其优化器也是完全自主开发的。主要包含3个组件：

1. Rewriter，以parse后生成的operator tree作为输入，做SQL-to-SQL的重写，rewrite可能基于heurstic或cost（这里是分布式cost），其中某些rewrite是bottom-up的，某些则是top-down的，不同rewrite之间可能会有交错。
   
2. Enumerator，类似传统的CBO，用来做access path的选择和join ordering，但这里会考虑分布式join ordering，并决定合适的数据分发方式。此外，rewriter在做cost based rewrite时，会调用enumerator接口来计算cost。
   
3. Planner，将产生的plan转换为SQL-like的子query以及对数据转移的描述。也就是说，subplan的分发是以SQL的形式，而不是传输物理plan。
   

Rewriter

可以基于heuristic或cost，要计算cost时要调用enumerator计算分布式cost，rewrite可以是bottom-up/top-down的。

1. 某些rewrite总是有益的，例如Column Elimination，总是可消除不必要的列从而减少disk/network IO，减少cpu计算量，因此总是有益的，一定可以应用。
   

2. 还有些rewrite基于启发规则，例如subquery merging，虽然不一定总是最优，但由于提供了更多的join order和join方式的选择，且在分布式环境下计算subquery成本往往更高，因此可能展开会更优，但另一方面，如果join中涉及到几个view，而每个view都涉及若干的表，全部展开可能会极大扩展搜索空间，此外，原有view中存在的一些语义关系（例如star schema join)也会被展开所破坏，反而不如不rewrite，而是各个view内部高效优化，再在外层做bushy join来的高效，因此需要有一些限定条件，来约束是否做rewrite。

此外，由于各种rewrite之间是会有相互影响的，因此需要交错进行，例如一个多层嵌套的left outer join，可以利用where条件中，内表列上的null-rejecting，将其转换为inner join，而当outer->inner后，where条件中内侧的condition就可以下推到内侧上，再次帮助判断内侧是否可以outer->inner，通过交错apply。

3. 另一些rewrite不确定其是否更优，例如 Group by Pushdown，也就是将group by下推到join的一侧中，将join -> group by变为group by -> join，那么显而易见，如果join后的cardinality小于join之前的表数据量(selective join)，那么后做group by需要的计算量更小，反之先做group by更节约计算资源，因此需要通过cost来比较转换前/后的plan，来决定是否应用。

在这里paper中强调了做costing时，一定是考虑分布式plan的cost。例如上图示例，如果是串行计划，可能由于join没有明显膨胀，下推group by的收益不高，计算cost会选择不下推，但如果考虑分布式，group by可以明显减少数据shuffle的network cost，很更可能可以下推。

在对rewrite做costing时，MemSQL也采用了和Oracle同样的思路，即对于计算过cost的query block，做某种cost标记，后续可以重用，此外每次只针对变化的部分重算cost，最小化costing overhead。

Handle Bushy Join

对于大量表做所有可能shape的join tree enumeration，成本是很高的，考虑到MemSQL还要枚举分布形态，search space就又多了一个维度，因此从设计决策上，在做Cost base join ordering时，它只枚举left-deep join tree。

但由于定位HTAP，MemSQL可能面临一些复杂的分析查询，其中包含一些星型/雪花模型的数query，而这种query对于bushy join的依赖是比较强的，因此MemSQL的设计决策是，在rewrite期间，基于heuristic检测可以做bushy join的pattern，并对每种潜在可能进行rewrite+costing，然后基于cost比较各种候选bushy plan的最优性。

为了是left-deep的enumerator可以枚举bushy tree，需要将bushy的部分rewrite为derived table，这样enumerator就会对各个query block各自优化，再join，在每个bushy的内部，仍然是left-deep的。

算法本身也并不复杂

1. 遍历所有join tables，形成join graph，其中点是table，边是join关系
2. 识别候选的satellite table集合:  表上有single table predicate
   例如column = constant  or column IN (constant,…) ，且表只和一个table相连
3. 识别候选的seed table集合:  表与至少两个其他表相连，且其中至少有一个是satellite table
/* 注意： 从2-3两个条件可以看到任意候选satellite table只能和一个seed table相连 */
4. 对每个候选seed table，依次执行如下步骤:
   4.1 将尽量多相连的satellite tables(dimension table)包含进来，形成derived table
   4.2 将所有可以推入的predicates,推入derived table
   4.3 在derived table内做column elimination
   4.4 形成一个候选plan Ci，调用enumerator计算这种join的cost，如果cost更低则保留，否则丢弃

paper中以TPC-DS Q25为例：

SELECT …
FROM 
store_sales ss,
store_returns sr,
catalog_sales cs,
date_dim d1,
date_dim d2,
date_dim d3,
store s,
item i
WHERE 
d1.d_moy = 4
AND d1.d_year = 2000
AND d1.d_date_sk = ss_sold_date_sk
AND i_item_sk = ss_item_sk
AND s_store_sk = ss_store_sk
AND ss_customer_sk = sr_customer_sk
AND ss_item_sk = sr_item_sk
AND ss_ticket_number = sr_ticket_number
AND sr_returned_date_sk = d2.d_date_sk
AND d2.d_moy BETWEEN 4 AND 10
AND d2.d_year = 2000
AND sr_customer_sk = cs_bill_customer_sk
AND sr_item_sk = cs_item_sk
AND cs_sold_date_sk = d3.d_date_sk
AND d3.d_moy BETWEEN 4 AND 10
AND d3.d_year = 2000
GROUP BY …
ORDER BY …

其join graph如下：

只考虑left-deep tree可以得到的最优plan和bushy tree可以得到的最优plan如下

可以看到left-deep join中，红色框图的join是成本较高的，由于d3只能cs有join condition，导致这个join是笛卡尔积。

而从join graph中可以识别出的候选satellite table集合包括{d1, d2, d3} ，候选seed table集合则是ss (connected to satellite d1 and sr), sr (connected to satellite d2 and ss), and cs (connected to satellite d3 and sr)，heuristic bushy算法会考虑每种候选的derived table组合，最后选中的最优plan是(d1, ss, sr, d2, s, i, (d3, cs))，如上图(b)所示，其所对应的示意SQL为

SELECT …
FROM store_sales,
store_returns,
date_dim d1,
date_dim d2,
store,
item,
  (SELECT *
   FROM 
   catalog_sales,
date_dim d3
WHERE cs_sold_date_sk = d3.d_date_sk
AND d3.d_moy BETWEEN 4 AND 10
   AND d3.d_year = 2000) sub
WHERE ...

这个方案的思想非常有意思，从算法中可以看到，选择bushy tree时，完全没有考虑cardinality 或任何统计信息，只是从join graph的语义出发，这样倒是可以避免统计信息不准确带来的问题。既可以大大减少plan space的膨胀，也可以利用到bushy join对于star join的高效性，对于像MySQL这种也只枚举left-deep tree的optimizer，可以做相对简单的改造就能实现bushy tree的引入。

不过从算法看，约束还是比较严格的，应该只能适用于具有严格形态的star join，此外仅给出了一个比较含糊的real world示例，不太确定该算法真实的效果。

Enumerator

接受Rewriter生成的候选operator tree作为输入，进行基于cost的物理优化生成best plan并进行标注，将带有标注的physical operator tree传入Planner。

1. 使用了类似SystemR的bottom-up DP机制，但由于是分布式系统，扩展了interesting order => interesting property，加上数据分布信息来计算cost。
   
2. enumerator本身会从底向上，依次对每个query block，独立优化（不会交叉），每个query block内部，只做left-deep的join enumeration。
   
3. 本身的cost计算是考虑分布式代价的(数据传输cost)，且cost model可以兼容rowstore和columnstore，列举了exchange的网络IO cost公式
   

Broadcast : R*D 

Repartition cost = (1/N) * (R * D + R * H)

R是传输行数，D是单行网络传入cost，H是单行hash evaluation cost。

Planner

Planner 将标注后的operator tree生成真正的物理执行计划，并划分为若干的DQEP steps，每个step用SQL来描述，这样比较容易扩展，而且SQL分发到各个leaf node后，可以在其本地利用更多的优化机制（例如compilation，plan cache...）。

Planner中引入了Remote Table和Result table这两个SQL extentsions，分别用于描述跨节点数据分发以及本地SQL计算。

Remote Table

用于表示leaf node获取其他leaf nodes的数据，例如：

SELECT facts.id, facts.value
FROM REMOTE(facts) as facts
WHERE facts.value > 4

上面SQL中的REMOTE关键字表示需要从facts table的所有partition中获取数据。

Result Table

MemSQL会为查询的中间执行结果在每个leaf node本地定义对应的result table，来存储这个分布式的中间数据，供后续执行使用。例如：

CREATE RESULT TABLE facts_filtered
AS SELECT facts.id, facts.value
FROM facts
WHERE facts.value > 4

以repartition为例举一个两者配合使用的例子：

(1) CREATE RESULT TABLE r1 PARTITION BY (y.a) AS
    SELECT * FROM y WHERE y.c > 5 (on every partition)
(2) SELECT * FROM x JOIN REMOTE(r1(p)) 
    WHERE x.b < 2 AND x.a = r1.a (on every partition)

步骤1先对y表数据，按y.a做repartition，分发到每个leaf node上。

步骤2对x表的每个本地分区数据，和y做分发后的数据，做分布式join。

注意这里的result table只是逻辑概念，并不一定是把数据物化下来，也就是说，和SQL Server PDW不同，plan是允许流水线的。

总结

本篇paper介绍了MemSQL新一代的查询优化器，其扩展了原有的2-pass logical -> physical的优化模式，使得transformation可以用cost来度量。

此外，它强调应该针对分布式环境进行planning，而不是走类似SQL Server PDW或CockroachDB那种先枚举最优串行计划，再做分布式优化的2-pass方案，这样可以获取更优解。

另一个个人觉得可以借鉴的就是heuristic based bushy join，对于MySQL这类不具备bushy join tree能力的优化器来说，可以做一些深入调研，看看这种相对来说实现成本较低的方案，是否具有可行性。