calcite 概念和架构

1. 前言   

        Flink使用Calcite构造SQL引擎，那么他们 是怎么合作的？ drill, hive，storm 和其他的一干apache 大数据引擎也用calcite ， 那么对于同一个sql 语句（statement) , 无论复杂简单与否，他们和Flink产生的执行计划是不是一样的? 如果不一样, 区别是怎么产生的？ 应该在哪里实施优化和发力？优化的手段和原则有那些，等等？   本文不会对calcite 面面做具到的介绍，重点是SQL执行计划的优化框架，流程和策略， 对执行计划进行优化是calcite 的主要业务。为了有助于理解优化框架，对于必要的概念会有介绍， 比如关系，关系代数，关系演算，等价原理，谓词逻辑等。

2. calcite 架构

 
  图-1 calcite 使用场景  
       如Calcite的官方论文 （见引用1， 发表于于 SIGMOD’18, June 10–15, 2018, Houston, TX, USA）所定义， calcite 是一个能够连接异构数据源，并运行优化的查询计划的基础框架 。Cacite 提供查询处理需要的三大功能：查询语言，查询优化和查询执行。 但是calcite 并不想做这样“one size fit all ” 的框架， 它提供足够灵活适配接口， 使外部系统 （数据源（或称存储系统），或查询处理系统）能够选择合适的场景与它适配 。calcite 支持的场景主要有两种， 如上图描述。
       在场景1中，calcite 作为独立运行的进程，后台通过适配器与外部的存储系统连接，前台通过JDBC 接口 使用SQL 语言通用户交互 。在这个场景里，calcite 作为一个中间件，为一些没有或缺乏友好的查询语言的存储系统（比如HBase, Cassandra, Kafka, ES, Redis） 提供查询语言（比如SQL）。 calcite在 内部将用户提交的查询优化并运行再自己的进程里，并在优化的过程中将适当的部分下推给存储系统 。比如Cassandra 有部分关系扫描（tableScan), 过滤（Filter), 投影（Proction) 和聚合（Aggregation）的能力， calcite 能将相应关系表达式翻译成Cassandra 的语言发送给它并返回结果。对于 Cassandra 不具备的能力，比如连接（ join），这部分关系表达式在calcite 里运行 。当然这个场景并不是calcite 擅长的，因为query 执行不是它所擅长的。
       在场景2中，calcite 作为嵌入式的组件运行在一个查询引擎里。这些查询引擎用自己的方式连接后台数据源，并使用自己的集群用分布的方式执行查询。  查询引擎善于获取数据和执行查询，需要calcite 提供查询语言和优化查询的能力 。 这个场景的例子有Hive , Flink, Drill , Storm 。 以 Flink 为例， 它的框架里有算子连接各种的异构数据源用于数据的获取和发布，无论是数据流还是数据集。  它也有丰富的算子将讲这些数据过滤、放大、缩小和变形用于各种各样计算需求。他需要一种对用户友好的， 对于批流数据语义统一的查询语言，以便于用户编写查询、将查询优化称执行计划和生成物理作业图， 使作业能够以最小的代价运行在Flink集群中。 场景2是calcite 最受欢迎的和最擅长的场景， 相比查询执行, 连接数据源， calcite 更擅长的是制造查询语言，解析查询， 和查询优化 。为什么这么说呢 , 那就要看看 它的架构了。  
 
图-2 引用1中calcite 官方论文中的架构图。        绿色框框QueryOptimizer （优化器， 也称Planner, 比如HepPlanner, VolcanoPlanner)是calcite 的心脏和大脑，它接受查询计划，输出优化的查询计划 。      蓝色的框框是优化器的输入和输出和各种适配器，包括Opeator Expressions (输入的原始计划，中间结果和最后输出的计划）, MetadataProvider(提供元数据的组件，比如对优化规则用的统计信息（RowCount of table, Distinct RowCout/min/max of a column , etc )  还有pluggable Rules (优化规则, 利用关系代数或关系演算的等价关系，优化执行计划，使之更够最快速的执行)。这三个组件是calcite 可扩展部分，因此与外部系统有连接 。      黄色框框（Data Processing System， 简称DPS)与蓝色框框有虚线连接，是DPS 对calcite 的扩展部分。  这里的
Data Processing System所指的就是场景2里的查询引擎。它通过扩展metadata provider 和 pluggable rules , 向优化器提供更准确的元数据信息，更适合的代价模型， 更高效的优化规则， 利用calcite 优化器产生最优化查询计划。SQL parser and validator, 是Calcite的SQL 语言的解释器， 它将用用户用SQL语言编写的查询解析称Opeator Expressions ， 并验证它的合法性 。  
       Opeator Expressions是一种用于表示关系代数表达式的树状数据结构。解释器将SQL 查询解释成关系代数表达式， 之后优化器调用规则将其修改为最优表达式。优化规则会根据有关系代数的等价原理将表达式变形从而使表达式的代价降低。但如何判断代价是否降低？ 办法有两种：一种是根据经验，一种是根据代价模型。根据经验学名称做启发式（Heuristic )模型, 根据代价模型估计的，学名叫火山模型。 相应的优化器也被称作启发式优化器和火山优化器（HepPlanner, VocanoPlanner)。         代价模型的量化计算是根据从metadata provider获取关系及关系运算的元数据，再辅以量化模型的计算。元数据通常指的是一个关系（table）和关系运算(projection, filter, join, aggregation, etc)产生关系的统计数据：如关系的row count, 某个分量的 distinct count， min, max 等 。DPS 会扩展Calcite 逻辑关系表达式产生“物理”关系表达式, 而这些扩展的表达式也会输入给优化器， 利用规则继续优化 。 
        Expression Builder 是一种绕过SQL解析，直接生成关系表达式的工具。 这种方式适用于单元测试，就不做展开展开介绍了。
 

图-3 calcite 交互数据流
         用数据流图的方式描述一下calcite 和Data Processing System 的交互， 如上图所示， 应该更容易理解 ， 为什么一个大数据处理系统更喜欢在场景2里面使用calcite 。优化器是从calcite的核心是不变的地方。 Rules, metadata, 和operator 是优化器依赖的帮手， 但都是可扩展的部分  。 回答在前言里的问题， 有那么多的大数据系统使用calcite， 相同的sql 查询经过calcide 优化器产生的计划会一样吗 ？ 答案是否定的。 原因是要看各个大数据系统里扩展的Operators, metadata provider , 还有新的规则和应用的顺序 是否相同等等 。不同的扩展，产生的关系表达式肯定不一样。 扩展做的更优秀， 优化的结果就更优秀。 比如Metadata Provider提供的关系运算统计数据是一种估计， 除了最底层的TableScan的代价估计是相对准确的， 别的运算的估计都是有误差的 。 误差越小，规则运行的越准确， 反之则不然 。    

3. 一些关系的概念

Calcite 只支持于关系型数据模型（不支持层次，网状，对象数据库的模型）， 那么什么是关系型数据库呢 ？ 建议读一下引用5中的那本书 ，虽然我也没读完 。下面解释一下一些比较容易混淆的概念 。

关系：关系一词来自离散数学里的集合论，根据维基百科的的定义，给定任意集合A和B，若 R属于AXB（笛卡尔乘积），则称R为从A到B的二元关系，特别在A=B时，称R为A上的二元关系。如果一个有N列的二维表， 每一列的取值范围为Ai  则该表是定义在A1x..Ai..xAn上的N元关系。可见，关系（Relation)是N元有序序列的集合。关系在数据库的概念里称作表（Table），关系的每一个有序序列叫做元组或行（Row),  元组的每一个量叫分量或列(Column)。

关系模式（Schema）：是对关系或表的描述。包括关系名称（Table Name），列名以及列的定义域（Domain)。

关系模型（Model)：指的是一系列关系模式的集合， 概念上对应数据库。

维度（Dimension)：通常是指列离散定义域的列。定义域上的每一个值称为基（cardinality), 一个关系已经使用的所有的基的个数成为基数 （cardinal number) , 也就是 distinct count , 也成 NDV （Number of Distinct Value) 。

关系代数：是由 Edgar F. Codd提出一种利用具有良好语义的代数结构用于对数据建模和定义查询的理论。代数结构是在一种或多种运算下封闭的一个或多个集合，那么关系代数在闭合关系上的良好语义的运算的集合。通俗的说，关系代数是一种通过由代数运算和输入关系组成的表达式来表达输出关系的理论。 比如 图-3中，SQL 产生的关系可以多个树形的表达式表示，树的形状和节点的排列顺序代表运算的过程（从下到上） 。关系代数是面向过程的。
关系演算，是另外一钟表达关系的理论， 他是以数理逻辑中的谓词演算为基础，用描述和声明的方式表达关系 。比如关系代数表达式用一系列代数运算来表示最终的产生关系，而关系演算则用声明形式的表达式描述一下最终的关系定义 。每一个关系代数的表达式，都有对应的等价的关系演算表达式（Codd定理）， 但反之则不然。关系演算和关系代数都来自关系模型理论。
SQL语言：是关系代数和关系演算的实现。 比如运算都来自关系代数， 运算里谓词都来自关系演算 。 现在大多数人都说SQL是一种声明式的语言，有点道理但也不全对。演算部分是声明的，代数部分是过程的。 比如一个equal join , join 是代数规定结果是一个笛卡尔乘积。 equal 是演算定义了目标关系里保留的是双方的键值要相等。 这个声明给了关系表达式的实现者可以通过hash join, 或sort merge join 优化 这个equal 。
Operator Expression : 指的是表达关系代数和关系演算的表达式， 在calcite 中由一种树状结构来表示 。它取这个名字的原因是树中所有的节点都有相应的操作符来表示， 包括输入和输出关系（TableScan， Sink )。我更喜欢把它叫做关系表达式：表达关系代数和演算的表达式 ，或由作用在关系上的运算组成的表达式。在优化的过程中，最原始的树状的数据结构会转化成一个图， 因为一些子计划可以重用的，重用的子计划和原计划会使用相同的节点实例，就像一个可以重入的函数可以被调用多次，但返回的数据是相同的。一个节点有多个parent, 就变成了不在是一棵树了。  比如SQL中的CTE， 如果上游的谓词没有下推，它是一个非常独立的存在，可以单独优化，形成一个比较独立的子计划和子表达式。　如果该子表达式被使用多次，他在图中就会成为多个上游节点的子计划。物化视图也是一个例子。 计划图是一个有向无环图(DAG)，也就是一个关系只能作为另外一个关系的输入，不能作为自身的直接或间接输入，树是有向无环图的特殊性形式，所以把优化后关系表达式称为计划图比较合适　。
还有一点要注意，当我们谈论计划的时候， 计划图是一个向下生长的DAG, 根节点是Sink, 叶结点是TableScan 。 上游，下游， 上推， 下推， 是对应从根节点先下的方向。当我们谈论作业流（jobGraph)的是，上游，下游对应的是数据流向。

4. calcite的概念

       上一章的概念说了这么多， 感觉就是为了解释什么是Operator Expression（关系表达式） 。这个概念很关键， 它是优化器操作的数据结构， 它经过优化规则的修剪和雕琢， 和元数据提供者营养的滋润，成为最终的最优（代价最小）结构。如果还是不好理解， 那就先抛去那些虽严谨但晦涩的数学概念， 可以把一个关系表达式想像成一个作战计划， 比如著名的官渡之战。 如果曹操在官渡和袁绍死磕, 作为原始的作战计划，也有可能也会获胜，毕竟有官渡河天险可守。但是代价会很高，毕竟袁绍的兵力数倍于曹操，正面作战胜率较低。于是曹操优化了原始计划，首先不主动出击，坚守官渡，然后 偷袭乌巢，烧了袁军的粮草，导致袁军军心大乱，仓促攻曹，最后致败。曹操用最小的代价大败袁绍于官渡，他的作战计划里的关键步骤是奇袭乌巢和坚守官渡。可惜袁绍白白拥有10万大军，和五子良将张颌，占尽优势，但却一败涂地。袁绍最大的错误就是使用了错误的计划， 信任千古奸人 郭图，这个人不仅害袁绍， 袁的两个儿子也都被他害死，最后被曹操斩杀 。不得不说，曹操无论是对官渡之站的计划，还是对郭图的计划都是最优的。
     如果做一个粗糙的类比，曹操的大脑在相当于calcite 的启发式优化器，他手下谋士的计策就是Plugable rules ，比如荀攸的突袭白马，荀彧的坚守官渡，许攸的奇袭乌巢，进攻、退守是关系代数里的运算符， 白马，乌巢，官渡这些地名相当于关系，那么行军路线就构成了一个计划图，如同关系表达式的计划图一样。作战计划依赖主将的智商和经验判断来计划的优劣，曹操雄才大略是不世的英雄豪杰，比起袁绍要聪明数倍， 他的判断自然准确率比较高。但是也有判断失误的时候啊，比如赤壁之战，依赖智商并不总是一个好办法。Calcite 的优化器要依靠元数据提供者的数据和代价模型用量化的指标来判断不同计划的优劣，在统计上应该是更准确的。那么回到一个程序员的思维里，优化器具体是怎么工作的呢 ？
先看下callcite 对外开放的接口。  
 
图-4 来自引用7中， Calcite API and SPIs

概念比较多， 用一个表来解释一下吧。

//build program
val builder = new HepProgramBuilder()
builder
  .addMatchLimit(10)
  .addRuleInstance(SubQueryRemoveRule.FILTER)
  .addRuleInstance(SubQueryRemoveRule.JOIN)
  .addMatchOrder(HepMatchOrder.BOTTOM_UP)
val hepProgram = builder.build()
//create planner
val planner = new HepPlanner(hepProgram, ...)
//build new operator expression graph
planner.setRoot(root)
planner.changeTraits(desiredTraits)
//apply rules in programs to optimize the operator expression
planner.findBestExp

5. 优化器流程

HepPlanner 的寻优流程很简单，setRoot重建planGraph, findBestExp 就是按照指定的顺序将HelpProgram里的规则触发一遍。 如果担心有问题贪心算法的问题，可以将这两步多做几次。

VocanoPlanner 的寻优流程如前所述。 TraitSet 通常包含了 Convension, distribution,  collation 三个维度 ， 这三个维度不同的基的组成的组合（subset)都是等价的但是cost不相等，但并不是代价把最低的subset 输入给上游总代价就会最低的。最低的代价需要综合考把虑上游和下游的情况，寻找最搭配的搭档。所以这个寻优过程需要一个动态规划的方式来求解。在使用动态规划（也就是递归的方法, volcanoPlanner 的命名就来自这里吧 ）求解之前， 我们需要把各种可能的计划的每一层的满足需要trait的subset， 以及对应关系求出来。 而满足要求的subset, 而且在考虑到输入的组合，状态转移公式大概如下。

 
BestExp(subset)) =  argmin( cost(rel1 ), cost(rel2), ... )
cost (rel ) = algoCost(rel.self) + cost(BestExp(rel.input1)) + cost(BestExp(rel.input2)) 

　第一行中，rel1, rel2  是满足traits 的等价表达式，表示当前subet的最佳表达式是他们之中里cost最低的表达式。

   第二行中， 表示表达式的cost 等于最上层节点自身算法代价估计 和下层输入subset的最佳表达式向上输入的累计的综合代价 。TableScan 的下层输入就是磁盘IO的代价，底层关系的RowCount相关 。这里的加号代表代价计算要综合考虑的因素并不是简单的算数相加 。比如hashJion和sortMergeJoin自身算法的代价不一样（CPU， 内存代价）， join的两路输入方式不同代价不一样（IO代价）。

 
举个例子 ：

SELECT * from store_sales join date_dim on store_sales.ss_sold_date_sk = date_dim.d_date_sk and date_dim.d_year=2002

这个表达式里主要有一个HashJoin 和两个TableScan 组成， 如果对HashJoin 的 requriedTraits 是 BATCH_PHYSICAL.ANY.[], 
HashJoin 对store_sales 和 date_dim的要求分别是BATCH_PYSICAL.forward.[] 和 BATCH_PYSICAL.broadcast.[], 那么上面的公式以下的形式 。
 

BestExp(join.BATCH_PHYSICAL.ANY.[]) = argmin( cost(shuffledHashjoin), cost(broadcastHashJoin), cost(shuffledSortMergeJoin)，... )
cost(broadcastHashJoin)= cost(BestExp(tablenscan_store_sales.BATCH_PHYSICAL.forward.[])) + cost(BestExp(tablenscan_date_dim.BATCH_PHYSICAL.broadcast.[])) + algoCost(HashJoin)

类似的过程可以求出 cost(broadcastHashJoin), cost(shuffledSortMergeJoin) 。
Broadcast join 极大的减少了网络和磁盘IO， cost 肯定是最低的 ， 最终Broadcast join表达式会选为join 层满足要求的subset 的最佳关系表达式 。
  从上面的转移公式可以看到， required traits 是从通常从上向下传递的（也有自我要求的) ，比如join对下游的广播方式的要求， 全局排序要求下游以single 方式发布数据的要求等。   cost是从下先上传递的， 没有下游已经确定的关系，上游是无法计算代价的 。在向下传递traits 的过程中， calcite 创建AbstractConverter代表 目标traits 临时节点， 之后再ConventionExpansionRule 将
AbstractConverter 转化成实际的关系。比如AbstractConverter.Broastcast.[] 建在TableScan上面， 目的是想让tableScan以广播方式输出， ConventionExpansionRule最终会将这个表达式变成BroadcastExchange+TableScan 。当TableScan的代价估计会沿着Exchange向上传递， 上游关系的代价也得以计算， 以此类推 。 VolcanoPlanner 将与节点匹配上的 
ConventionExpansionRule 和其他的ConverterRule 都放在优先队列里。由于新的relSubset创建时， AbstractConverter才会创建，之后触发ConventionExpansionRule 与之匹配和放入队列， 用以之后创建Exchange 和 sort 节点。这个和其他的ConverterRule 不同， ConverterRule 是匹配旧的Convention 的结点（比如LogicalXxxxx), 他们在节点注册的时候（setRoot)就已经入队。优先队列里的成员都有优先级，级别高的先被调用， 这样能保证那些ConverterRule 先于ConventionExpansionRule 调用。  
 

图-5 VolcanoPlanner

画个图表示一下 relSet, relSubset, 和besExp 还有trait之间的关系吧， 还有这个类似火山喷发的形状。

7. 引用