calcite连接mysql_如何使用Calcite实现一个简单的数据库

说道Calcite你可能有些陌生&＃xff0c; 但提及Hive、Kylin、Apache Drill、Flink等一定不会陌生&＃xff0c;这些都是在我们日常工作中经常用到的&＃xff0c;如上这些都是基于Calcite实现查询引擎&＃xff0c;还有Druid和Storm也是使用它来实现SQL功能。按照官方的说法&＃xff0c;Calcite是动态数据管理框架&＃xff0c;这个解释理解起来有点抽象&＃xff0c;通俗一点讲&＃xff0c;要使用Calcite实现数据库&＃xff0c;只需要关注存储引擎以及元数据管理&＃xff0c;其他都交给Ca

lcite。可能这个说法有些不严谨&＃xff0c;文档中还提到了Calcite不提供处理数据的算法&＃xff0c;但Calcite-core和Calite-linq都提供了一些算子的实现对于一个简单的数据库足够了。

对于Calcite的详细介绍推荐大家看一遍文章&＃xff0c;本文主要介绍Calcite如何使用&＃xff0c;例如已经有一种数据格式的文件存储&＃xff0c;如何利用Calcite快速实现SQL查询。我看过kylin、druid的Calcite应用&＃xff0c;也是各不相同&＃xff0c;这大概也是Calcite的魅力吧。

Calcite文档有一个指南&＃xff0c;介绍使用CSV File作为数据存储格式实现SQL查询&＃xff0c;掌握了以后我们可以照猫画虎造出一个其他数据格式的数据库&＃xff0c;或者对学习kylin、druid的源码有帮忙&＃xff0c;概括地讲&＃xff0c;在这个例子使用以下技巧&＃xff1a;

自定义Schema

自定义Table

决定Table的字段类型

使用ScannableTable实现简单的全表扫描

更高级一点的技巧&＃xff0c;使用Filterable Table实现谓词下推

更酷一点的技巧&＃xff0c;基于TranslateTable使用Rule实现逻辑表达式的转换

前四点是构建一个简单的&＃xff0c;采用全表扫描的方式实现查询。5和6属于进阶内容&＃xff0c;在案例中&＃xff0c;使用Rule转换的方式实现了Project下推&＃xff0c;和5实现的谓词下推是常用的SQL优化方式。下面由浅到深介绍这几项技巧。先来看前四项&＃xff0c;完成一个简单的只能全表扫描的数据库。

首先在GitHub上下载Calcite的源码&＃xff0c;看calcite-example-csv工程&＃xff0c;在src/test/CSVTest中有各种场景的测试用例&＃xff0c;例如

testFilterableWhere是测试谓词下推

testPushDownProject是Project下推

testSelect是最简单的全表扫描

可以先跑一下测试用例感受一下Calcite的魅力&＃xff0c;Calcite实现一个数据库&＃xff0c;只需要关注存储引擎以及元数据管理。存储格式采用csv&＃xff0c;一个CSV文件会映射成一个Table&＃xff0c;需要注意的是CSV文件的第一行是Table的元数据信息&＃xff0c;采用“FieldName1:FieldType,FieldNameN:FieldType”这样的格式存储&＃xff0c;类似excel中的表头信息。以下是sales/SALES.csv的示例。

DEPTNO:int,NAME:string

10,"Sales"

20,"Marketing"

30,"Accounts"

至此介绍了存储格式以及元数据&＃xff0c;接下来介绍如何使用。

第一步&＃xff0c;创建一个json格式的mode文件&＃xff0c;描述了如何创建Schema&＃xff0c;可以参照test/resource目录下的model.json,

{

"version": "1.0",

"defaultSchema": "SALES",

"schemas": [

{

"name": "SALES",

"type": "custom",

"factory": "org.apache.calcite.adapter.csv.CsvSchemaFactory",

"operand": {

"directory": "sales"

}

}

]

}

在分析model文件之前&＃xff0c;我们先了解几个重要的概念&＃xff1a;

Schema&＃xff0c;是table和function的名称空间&＃xff0c;它是一个可嵌套的结构&＃xff0c;Schema还可以有subSchema&＃xff0c;理论上可以无限嵌套&＃xff0c;但一般不会这么做。Schema可以理解成Database&＃xff0c;Database下面有table&＃xff0c;这样就和传统数据库的概念联系起来了&＃xff0c;在Calcite中&＃xff0c;顶层的Schema是root&＃xff0c;自定义的Schema是root的subSchema&＃xff0c;同时还可以设置defaultSchema&＃xff0c;类似我们使用数据库时&＃xff0c;使用use database命令以后就不用再输入database名字前缀。

Table&＃xff0c;就很好理解了&＃xff0c;就是数据库中的表。在table描述了字段名以及相应的类型、表的统计信息&＃xff0c;例如表有多少条记录等等&＃xff0c;这里先不展开讲。另外重要的是数据文件的存储以及如何扫描读取数据文件。

那么再去看这份model文件&＃xff0c;就比较清晰明了。它描述了在数据库中有多少个Schema、每个Schema如何创建以及默认的Schema&＃xff0c;这里的Schema可以理解成database。defaultSchema属性设置默认Schema&＃xff0c;schemas是数组类型&＃xff0c;每一项代表一个Schema描述信息&＃xff0c;在描述信息中有一个关键的属性factory&＃xff0c;它是创建Schema的工厂类&＃xff0c;在这个例子中factory是org.apache.calcite.adapter.csv.CsvSchemaFactory&＃xff0c;它实现了SchemaFactory接口。

要自实现只有全表扫描功能的简单数据库需要做如下几步&＃xff1a;

自定义SchemaFactory

自定义Schema

自定义Table

自定义Enumerator

先看看SchemaFactory接口&＃xff0c;它只有一个方法&＃xff1a;

Schema create(

SchemaPlus parentSchema,

String name,

Map operand);

create用于创建Schema&＃xff0c;其参数说明如下&＃xff1a;

parentSchema&＃xff0c;他的父节点&＃xff0c;一般为root

name schema的名字&＃xff0c;它在model中定义的

operand&＃xff0c;也是在mode中定义的&＃xff0c;是Map类型&＃xff0c;用于传入自定义参数。

在这个Model中&＃xff0c;CSVSchemaFactory创建一个叫“SALES”的CSVSchema&＃xff0c;它会把src/test/resources/sales下所有CSV文件构建成table。所以operand只许设定了一个参数directory&＃xff0c;即读取CSV文件的根目录。CSVSchemaFactory的实现比较简单所以就不在展开分析&＃xff0c;需要注意是的源码中flavor参数的处理&＃xff0c;这个参数涉及优化进阶相关&＃xff0c;这里先不用管&＃xff0c;默认为SCANNABLE。

自定义Schema需要实现Schema接口&＃xff0c;前面提过Schema是table和function的名称空间&＃xff0c;其主要方法如下&＃xff1a;

Table getTable(String name)&＃xff0c;根据表名获取Table

Set getTableNames()&＃xff0c;获取Schema下的所有表名集合

Collection getFunctions(String name)&＃xff0c;根据函数名获取函数列表&＃xff0c;和table不同&＃xff0c;这里返回的是集合类型。

Set getFunctionNames()&＃xff0c;或者所有的函数名集合。

CsvSchema->AbstractSchema->Schema,AbstractSchema重新设计了一个getTableMap方法&＃xff0c;使用tableName->Table的Map结构存储所有table。这样设计的好处是getTable()能够快速查找。CSVSchema的实现也比较简单&＃xff0c;遍历读取根目录下的每个

文件创建成表&＃xff0c;因为上面的model.json中flavor没有设置&＃xff0c;采用默认值SCANNABLE&＃xff0c;创建成CsvScannableTable。

自定义Table是本文中最复杂的&＃xff0c;先看下图&＃xff1a;

BOOLEAN(Primitive.BOOLEAN),

BYTE(Primitive.BYTE),

CHAR(Primitive.CHAR),

//只列举部分类型

由如上代码可知&＃xff0c;type并不都是标准的SQL Type&＃xff0c;例如String。Calcite中设计了RelDataTypeFactory&＃xff0c;不仅支持标准的SQL TYPE&＃xff0c;也支持java类型以及Array、Map等集合类型。该实例中&＃xff0c;RowType是一个StructType&＃xff0c;是集合类型&＃xff0c;类似c语言中的struct&＃xff0c;非常适合存储行记录中字段名以及类型&＃xff0c;这和Hive的方式是一样的。例如SALES文件中的

DEPTNO:int,NAME:string

则Type为

struct

在这个例子中通过读取csv文件的第一行来获取fieldName以及fieldType的&＃xff0c;具体实现在CsvEnumerator的deduceRowType()方法。

在calcite中一般有两种执行模型&＃xff0c;解释和编译&＃xff0c;这一点类似Java。编译模式更好理解一些&＃xff0c;会把逻辑执行计划通过字节码技术生成java code然后编译执行。解释模式则省掉生成代码编译的过程。关于解释执行。我看过一些基于Calcite的应用&＃xff0c;大部分还是采用编译模式的&＃xff0c;所以你看完这篇文章以后再去看其他使用calicite的项目&＃xff0c;可能找不到熟悉的身影&＃xff0c;如果table实现了如下三个接口之一&＃xff0c;Calcite则会使用解释模式执行

ScannableTable

FilterableTable

ProjectableFilterableTable

ScannableTable用于简单的全表扫描&＃xff0c;FilterableTable用于谓词下推&＃xff0c;ProjectableFilterableTable更酷一些既能支持谓词下推又能支持project下推。他们都有一个scan&＃xff0c;但是参数不同

ScannableTable

Enumerable scan(DataContext root);

FilterableTable

Enumerable scan(DataContext root, List filters);

因为要做谓词下推&＃xff0c;比ScannableTable多了filters。filters是where语句中的filter。

ProjectableFilterableTable

Enumerable scan(DataContext root, List filters,

int[] projects);

又增加了projects&＃xff0c;投影字段顺序的数组。

Enumerable支持linq和java的迭代器

//返回java的迭代器

Iterator it &＃61; enumerable.iterator();

//LINQ风格的迭代器

Enumerator enumerator &＃61;enumerable.enumerator();

要使用这两种迭代器之前&＃xff0c;必须要实现它&＃xff01;AbstractEnumerable借助Linq4j实现了enumerator和iterator的转换

public Iterator iterator() {

return Linq4j.enumeratorIterator(enumerator());

}

所以我们仅需实现enumerator方法。

Enumerator是Linq风格的迭代器&＃xff0c;它有4个方法&＃xff1a;

current()

moveNext()

reset()

close()

current返回游标所指的当前记录&＃xff0c;需要注意的是current并不会改变游标的位置&＃xff0c;这一点和iterator是不同的&＃xff0c;在iterator相对应的是next方法&＃xff0c;每一次调用都会将游标移动到下一条记录&＃xff0c;current则不会&＃xff0c;Enumerator是在调用moveNext方法时才会移动游标。moveNext方法将游标指向下一条记录&＃xff0c;并获取当前记录供current方法调用&＃xff0c;如果没有下一条记录则返回false。

CsvEnumerator是读取csv文件的迭代器&＃xff0c;它还得需要一个RowConverter&＃xff0c;因为csv中都是String类型&＃xff0c;使用RowConverter转化成相应的类型。在moreNext方法中&＃xff0c;有Stream和谓词下推filter部分的实现&＃xff0c;在本文只关注如下几行代码&＃xff1a;

final String[] strings &＃61; reader.readNext();

if (strings &＃61;&＃61; null) {

current &＃61; null;

return false;

}

current &＃61; rowConverter.convertRow(strings);

return true;

至此&＃xff0c;我们完成了使用csv文件存储的数据库全部工作&＃xff0c;你可以在CsvTest中使用所有的名为“model”的模型进行测试&＃xff0c;

checkSql("model", "select * from EMPS");

//smart模型的会在后续的文中介绍

checkSql("smart", "select name from EMPS");

总结一下&＃xff1a;

创建模型,model.json

自定义SchemaFactory&＃xff0c;CsvSchemaFactory

自定义Schema&＃xff0c;CsvSchema

自定义Table&＃xff0c;CsvTable、CsvScannableTable

自定义Enumerator&＃xff0c;CsvEnumerator

分享的过程也是学习的过程&＃xff0c;在写本文过程&＃xff0c;也了解了不少以前自以为懂了的细节&＃xff0c;但也有可能还存在不正确的认识&＃xff0c;欢迎指正交流。微信号&＃xff1a;zhl5919

参照资料&＃xff1a;