Calcite技术研究

2020 年 2 月 24 日
笔记

原文作者：王长春，来自平安银行零售大数据团队

概述

Apache Calcite是一个基础的软件框架，它提供了查询处理、查询优化以及查询语言支持的能力。很多流行的开源数据处理系统例如Apache Hive,Apache Storm,ApacheFlink,Druid等都采用了它。

下图是采用Apache Calcite的开源数据处理系统，以及Calcite能连接到的数据源。

大多数数据处理系统是使用Calcite来做SQL解析和查询优化。还有部分使用Avatica(Calcite的子项目)来构建自己的JDBC driver。还有部分使用Calcite来重写查询请求以使用物化视图。

最近十几年来，出现了很多专门的数据处理引擎。例如列式存储、流处理引擎、文档搜索引擎等等。这些引擎宣称他们在特殊方面能提供更高“性价比”的性能，并且宣称“one size fits all”范式的时代已经终结了。Spark、Storm、Flink、Elasticsearch、Druid等多种引擎的流行确实说明了这一点。

这些数据处理引擎都面临着一些共同但是难以解决的问题。

一是数据处理引擎都要做查询优化以及提供sql查询语言或者其他DSL查询语言。
二是使用者可能使用了多个专门的数据引擎，例如使用了ES、Spark、Druid.那么使用者很可能会有在异构数据源上支持查询以及查询优化的需求。

Apache Calcite就是为解决这些问题而设计的。Calcite提供了所有数据处理系统所需要的查询执行、查询优化、查询语言等能力。但是Calcite没有提供数据存储以及数据管理的能力，这两个能力是由各自的数据处理引擎来提供的。此外，Calcite提供了跨平台查询优化能力。

总结一下，Calcite在现在这么流行，主要原因如下：

开源并完全按照Apache基金会的规则规范运作。Calcite已于2013年成为Apache顶级项目。
Calcite使用Java开发，便于被数据处理引擎集成。
Calcite数据模型强大，既能支持流式数据处理引擎，也能支持批式数据处理引擎。
Calcite优化器的每个模块都是可插拔的可扩展的，包括rules和成本模型。这使得Calcite优化器非常灵活。
Calcite能在多个数据处理引擎上执行查询以及做查询优化。
Calcite提供了ANSI 标准SQL，以及各种SQL dialect. 另外Calcite提供了符合JDBC 规范的Driver。

Calcite架构

Calcite包含了很多典型数据管理系统的组成模块。但是，它故意忽略了一些关键模块，例如数据的存储，数据处理算法，以及元数据的存储。但正是这些特点，使得calcite成为有多个数据存储和多个数据处理引擎的应用程序的中间层。当然，它也可以成为构建一个数据处理引擎的基础组件。

Calcite架构如上图所示。

Calcite优化器使用关系运算符树作为内部表现形式。优化引擎主要由规则、元数据提供者、以及planner engine组成。

Calcite包含一个查询解析器和验证器，可以将SQL查询转换为关系运算符树。由于Calcite不包含存储层，它提供了一种机制来通过适配器定义外部存储引擎中的table schema和views（适配器将在下文中详述）。这样，应用程序就可以通过calcite访问外部存储引擎。

在查询被优化后，calcite还能将优化后的关系表达式翻译回SQL。这使得calcite能够和有sql接口但是没有优化器的数据处理引擎很好的集成。

查询代数

3.1 运算符

Calcite的核心是关系代数。除了表达DML操作的基本运算符外，calcite还提供了一些额外的运算符，这些运算符能够简洁的表达复杂操作，或者能够更高效的识别优化机会。例如，OLAP中的decision making,以及流处理引擎中的窗口函数。Calcite引入了window运算符并且封装了window的定义，例如窗口的上下界、分区以及聚合函数。

3.2 Traits

Calcite不使用不同的实体来代表逻辑和物理运算符。它使用traits来描述运算符的物理属性。这些特质能够帮助优化器计算不同可选的plan的成本。改变特质的值不会改变已经计算过的逻辑表达式的值，举例：某个运算符已经产生的行数不会被改变。

在优化过程中，Calcite试图在关系表达式上强制执行某些traits，比如某些列的排序。关系运算符可以实现converter接口来指示如何改变表达式的traits的值。

Calcite包括一些通用特质，这些特质描述了关系运算符（例如ordering,grouping,partitioning）所产生的数据的物理属性。Calcite优化器能对这些属性进行推理和探索以发现不必要的运算符。例如如果sort运算符的输入已经是有序的，那么这个sort运算符就可以删掉。

除了通用特质，Calcite的一个重要功能就是calling convention特质。本质上来说，这个特质代表的是表达式所被执行的数据处理系统。这个特质使得calcite能够满足查询运行在多个引擎上时，优化可以透明地进行。

例如上图所示，mysql中有products表，Splunk中有orders表。理所当然的，orders表的scan是在splunk convention中。Products表的scan发生在jdbc-mysql convention中。Join是一个逻辑convention，即还没有实现被选择。另外，上图的sql查询还包括filter,这个运算符根据适配器的规则被下推到splunk。对join来说，一个可能的实现是使用Spark作为外部引擎。Join转化为spark convention，他的输入是从jdbc-mysql和splunk到spark convention的converters运算符。

适配器

适配器定义了calcite如何与各种数据源集成以访问各种数据源。适配器的组件如下图所示：

适配器由model、schema以及schema factory组成。Model是数据源的物理属性的规范。Schema是model中的数据的定义（主要是格式和布局）。数据本身可以通过table访问。Calcite使用适配层中定义的table接口访问数据。适配器可以定义一系列的规则并添加到planner中。例如，他可以添加以下规则：把逻辑关系表达式转化为这个适配层对应的引擎的关系表达式。Schema factory从model中获取元数据信息并生成schema。

章节3中提到过，Calcite使用calling convention来识别关系运算符属于哪一数据处理引擎。这些运算符为适配层中的表实现了访问方法。当查询被解析并且转化为关系代数表达式之后，calcite会为每个表创建一个scan的运算符。Scan运算符是适配层必须要实现的。如果适配层实现了scan运算符，calcite优化器就能使用client侧运算符例如sorting,filtering,join等。

适配器是一个很好的抽象，他使得查询优化不局限于某个数据处理引擎，可以跨多个数据处理引擎。Calcite可以把查询中涉及到的多个表逻辑下推到各自的数据处理引擎，然后再对结果数据执行聚合和join。实现适配器的最小要求是提供table scan 运算符，当然也可以包括更多高级的优化。任何关系代数中的表达式都能根据优化规则下推到适配器中。

Calcite本身已经实现了很多的适配器，当然我们也可以自己开发新的数据处理引擎对应的适配器。下图是calcite内置的适配器。

查询处理与优化

查询优化是calcite的一个重要功能。查询优化其实就是重复地在关系表达式上执行planner rule。这个过程依赖于成本模型，planner 引擎会试图产生一个有相同语义但是成本更低的替代表达式。优化器的各个组件都是可扩展的，你可以添加关系运算符，规则，成本模型，以及统计数据。

5.1 规划器规则（planner rules）

Calcite包括一系列的规划器规则来进行表达式树的转换。一个规则匹配树的一个模式并在不改变语义的前提下执行转换。Calcite包含几百个优化规则。当然也可以添加自己的规则到calcite中。

例如，Calcite提供了Cassandra适配器。Cassandra有以下特点：数据根据部分字段分区，且在每个分区中，行是根据另一部分字段排过序的。对适配器来说，下推尽量多的查询到backend中是非常好的查询优化。下推sort到cassandra的规则必须符合两个条件：

1）对表的查询过滤后只会到一个分区中（因为行在一个分区中是有序的）

2） Cassandra的分区排序和要求的排序有相同的前缀

若要符合这两个条件，需要把logicalFilter重写为cassandraFilter以实现分区filter 下推到cassandra。

再比如如下更复杂的查询：

SELECT products.name, COUNT(*)FROM sales JOIN products USING (productId) WHERE sales.discount IS NOT NULLGROUP BY products.name ORDER BY COUNT(*) DESC;

这个查询对应的关系代数表达式可以用下图中的左半部分表示：

因为where子句仅仅用于sales表，我们可以把filter挪到join之前如右半部分图所示。这个优化可以大幅减少执行时间。另外，如果这两张表不在同一个backend中，这个优化能使适配器把filter下推到backend中。这个优化就是calcite的FilterIntoJoinRule。

5.2 元数据提供者

元数据有两个重要作用：一是指导规划器降低整个查询计划的成本，二是提供信息给rules。元数据提供者负责提供这些信息给优化器。实际上，calcite的默认的元数据提供者实现包括这些功能：执行运算符树的子表达式的全部成本，行数，表达式的结果集的数据量，最大并行度。它还能提供plan结构信息，例如某个tree node下面有个filter condition。

Calcite提供了元数据plugin接口。数据处理引擎可以选择重写providers中已经存在的函数，或者提供他们新的元数据函数。在大多数情况下，数据处理引擎通过元数据提供者接口提供数据的统计数据（如行数数据量）就已经足够了，剩下的工作calcite通过它的默认实现就可以完成。

元数据提供者是可插拔的，他们通过一个java轻量级编译器Janino来编译和运行时实例化。元数据提供者的实现包括元数据结果的cache,这个cache能很大幅度提高性能。

5.3 规划器引擎

规划器引擎的主要功能是触发优化规则一直进行优化直到达到给定的目标。目前calcite提供两种不同的规划器引擎，新的引擎是可插拔的。

第一种是基于成本的规划引擎。它的规则的目标是减少整个表达式的成本。这个引擎使用动态编程算法，类似于Volcano。最初，每个表达式带着基于表达式的摘要以及输入注册到规划器。当一个规则在表达式e1上触发后，会产生表达式e2.规划器会把e2添加到e1所在的相等表达式集合Sa中。同时，规划器产生新表达式的摘要，这个摘要会和以前注册到规划器的摘要进行比较。如果一个相似的摘要（表达式e3的摘要）在Sb中发现，规划器发现了重复将会把Sa和Sb合并。这个过程会一直持续到规划器达到一个配置的固定点。规划器将穷举搜索空间直到所有的规则在所有的表达式上都应用过。或者当规划的成本在上个迭代中的改进小于某阈值时也会停止优化。元数据提供者需要提供cost函数，这个cost函数会被优化器用来决策哪个plan会被选择。默认的cost 函数会根据一个给定的表达式综合考虑CPU IO 以及内存资源。

第二个引擎是穷举式规划器。它会一直触发规则直到它产生的表达式不再被任何规则所改变。这个规划器在不考虑表达式成本的场景下是有用的，它可以快速执行规则。

用户可以根据自己的需求选择规划器引擎，也可以从一种引擎切换到另一种。或者，用户也可以产生多阶段优化逻辑，即在不同的阶段使用不同的规则集。

总结

本文主要描述的Calcite的架构以及基本原理，并简单介绍了Calcite的主要模块。Calcite的内容很多，后面还需要更深入的研究。