本章针对Calcite的连接建立提供了循序渐进的教程，使用一个简单的适配器来将一个CSV文件目录以包含Schema信息的tables形式呈现，并提供了一个完全SQL接口。
Calcite-example-CSV是一个Calcite中的一个功能完备的适配器，它可以读取CSV格式(以逗号分隔)的文本文件。值得称赞的是，几百行的java代码就足够提供完全的SQL查询功能。
CSV适配器同样作为一个其他数据格式的适配器构建参考模板。尽管代码量不大，但它覆盖了一些重要的概念:
1) 用户通过使用SchemaFactory和Schema interfaces来自定义schema
2) 以JSON模型文件声明schemas
3) 以JSON模型文件声明视图views
4) 通过Table interface自定义table
5) 定义table的record类型
6) 使用Scannable Table interface作为Table的简单实现，来直接枚举所有的rows
7) 进阶实现FilterableTable，来根据简单的谓词predicates过滤rows
8) 以Translatable Table进阶实现Table，将关系型算子翻译为执行计划规则
版本依赖：Java (1.7 or higher; 1.8 preferred), git and maven (3.2.1 or later).
$ git clone https://github.com/apache/calcite.git
$ cd calcite
$ mvn install -DskipTests -Dcheckstyle.skip=true
$ cd example/csv可以通过工程内置的sqlline脚本来连接到Calcite
$ ./sqlline
sqlline> !connect jdbc:calcite:model=target/test-classes/model.json admin admin(如果使用Windows操作系统，命令为sqlline.bat)
执行一个metadata 查询
sqlline> !tables
+------------+--------------+-------------+---------------+----------+------+
| TABLE_CAT | TABLE_SCHEM | TABLE_NAME | TABLE_TYPE | REMARKS | TYPE |
+------------+--------------+-------------+---------------+----------+------+
| null | SALES | DEPTS | TABLE | null | null |
| null | SALES | EMPS | TABLE | null | null |
| null | SALES | HOBBIES | TABLE | null | null |
| null | metadata | COLUMNS | SYSTEM_TABLE | null | null |
| null | metadata | TABLES | SYSTEM_TABLE | null | null |
+------------+--------------+-------------+---------------+----------+------+ JDBC提示：sqlline中的 !tables 命令实际上等于执行 DatabaseMetaData.getTables() , 还有其他命令来查询JDBC metadata, 例如 !columns 和 !describe。
如结果所示，该系统中共有5个table: SALES schema下的EMPS, DEPTS, HOBBIES 和系统自带的 metadata schema下的COLUMNS和TABLES。系统table在Calcite中会一直展示，但其他表是由schema的指定实现而来，在本例中，EMPS和DEPTS表来源于target/test-classes路径下的EMPS.csv和DEPTS.csv文件。
通过在这些表上执行一些查询，我们可以验证Calcite提供了完整的SQL功能实现。
首先，scan table：
sqlline> SELECT * FROM emps;
+--------+--------+---------+---------+----------------+--------+-------+---+
| EMPNO | NAME | DEPTNO | GENDER | CITY | EMPID | AGE | S |
+--------+--------+---------+---------+----------------+--------+-------+---+
| 100 | Fred | 10 | | | 30 | 25 | t |
| 110 | Eric | 20 | M | San Francisco | 3 | 80 | n |
| 110 | John | 40 | M | Vancouver | 2 | null | f |
| 120 | Wilma | 20 | F | | 1 | 5 | n |
| 130 | Alice | 40 | F | Vancouver | 2 | null | f |
+--------+--------+---------+---------+----------------+--------+-------+---+
同时支持JOIN和GROUP BY
sqlline> SELECT d.name, COUNT(*)
. . . .> FROM emps AS e JOIN depts AS d ON e.deptno = d.deptno
. . . .> GROUP BY d.name;
+------------+---------+
| NAME | EXPR$1 |
+------------+---------+
| Sales | 1 |
| Marketing | 2 |
最后，VALUES操作符可以聚合生成单独一行数据，我们可以通过这种简便的方法来测试表达式和SQL内嵌函数:
sqlline> VALUES CHAR_LENGTH('Hello, ' || 'world!');
+---------+
| EXPR$0 |
+---------+
| 13 |
+---------+
Calcite具有其他许多SQL特性。我们来不及在这里一一举例，使用者可以编写更多的查询来进行验证。
现在，我们来探索一下Calcite是如何发现这些table的。记住，最核心的Calcite不知道CSV文件的任何信息。（像一个“没有存储层的databse”一样，Calcite不会去了解任何文件格式）Calcite能识别这些table是因为我们告诉它去运行calcite-example-csv工程下的代码。
运行链中有一系列的步骤。首先，我们在一个schema 工程类中以model file的格式定义一个schema。然后schema工厂类创建一个schema，schema创建多个table，这些table都知道如何通过scan CSV文件来获取数据。最后，在Calcite解析完查询并将查询计划映射到这几个table上时，Calcite会在查询执行时触发这些table去读取数据。接下来我们更深入地解析其中的细节步骤。
在JDBC连接字符串中，我们会给出以JSON格式定义的model的路径。model具体定义如下
{
version: '1.0',
defaultSchema: 'SALES',
schemas: [
{
name: 'SALES',
type: 'custom',
factory: 'org.apache.calcite.adapter.csv.CsvSchemaFactory',
operand: {
directory: 'target/test-classes/sales'
}
}
]
}
这个model定义了一个名为SALES的schema。这个schema是由一个plugin类支持的，org.apache.calcite.adapter.csv.CsvSchemaFactory，这个类是calcite-example-csv工程里的一部分并且实现了Calcite中的SchemaFactory接口。它的create方法将一个schema实例化了，将model file中的directory参数传递过去了。
public Schema create(SchemaPlus parentSchema, String name,
Map<String, Object> operand) {
String directory = (String) operand.get("directory");
String flavorName = (String) operand.get("flavor");
CsvTable.Flavor flavor;
if (flavorName == null) {
flavor = CsvTable.Flavor.SCANNABLE;
} else {
flavor = CsvTable.Flavor.valueOf(flavorName.toUpperCase());
}
return new CsvSchema(new File(directory), flavor);
} 根据model的配置，这个schema 工程类实例化了一个名为SALES的schema。这个schema是org.apache.calcite.adapter.csv.CsvSchema的一个实例，实现了Calcite中的Schema接口。
一个schema的职责是产生一系列的tables（它也可以列举出子schema和table-function，但这些高级的特性在calcite-example-csv中没有支持）。这些table实现了Calcite的Table接口。CsvSchema生成了一些tables，它们是CsvTable以及CsvTable的子类的实例。
下面是CsvSchema的一些相关代码，对基类AbstractSchema中的getTableMap()方法进行了重载。
protected Map<String, Table> getTableMap() {
// Look for files in the directory ending in ".csv", ".csv.gz", ".json",".json.gz".
File[] files = directoryFile.listFiles(
new FilenameFilter() {
public boolean accept(File dir, String name) {
final String nameSansGz = trim(name, ".gz");
return nameSansGz.endsWith(".csv")|| nameSansGz.endsWith(".json");
}
});
if (files == null) {
System.out.println("directory " + directoryFile + " not found");
files = new File[0];
}
// Build a map from table name to table; each file becomes a table.
final ImmutableMap.Builder<String, Table> builder = ImmutableMap.builder();
for (File file : files) {
String tableName = trim(file.getName(), ".gz");
final String tableNameSansJson = trimOrNull(tableName, ".json");
if (tableNameSansJson != null) {
JsonTable table = new JsonTable(file);
builder.put(tableNameSansJson, table);
continue;
}
tableName = trim(tableName, ".csv");
final Table table = createTable(file);
builder.put(tableName, table);
}
return builder.build();
}
/** Creates different sub-type of table based on the "flavor" attribute. */
private Table createTable(File file) {
switch (flavor) {
case TRANSLATABLE:
return new CsvTranslatableTable(file, null);
case SCANNABLE:
return new CsvScannableTable(file, null);
case FILTERABLE:
return new CsvFilterableTable(file, null);
default:
throw new AssertionError("Unknown flavor " + flavor);
}
}schema会扫描指定路径，找到所有以".csv”结尾的文件。在本例中，指定路径是 target/test-classes/sales，路径中包含文件EMPS.csv和DEPTS.csv，这两个文件会转换成EMPS和DEPTS表。
{
version: '1.0',
defaultSchema: 'SALES',
schemas: [
{
name: 'SALES',
type: 'custom',
factory: 'org.apache.calcite.adapter.csv.CsvSchemaFactory',
operand: {
directory: 'target/test-classes/sales'
},
tables: [
{
name: 'FEMALE_EMPS',
type: 'view',
sql: 'SELECT * FROM emps WHERE gender = \'F\''
}
]
}
]
}上面的 type：“view”这一行将FEMALE_EMPS定义为一个视图，而不是常规表或者是自定义表。注意JSON中定义单引号需要加上转义字段“\”。使用JSON定义长字符串易用性不太高，因此Calcite支持一种替代语法。如果视图定义中有长SQL语句，可以使用多行来定义一个长字符串。
{
name: 'FEMALE_EMPS',
type: 'view',
sql: [
'SELECT * FROM emps',
'WHERE gender = \'F\''
]
}在定义完一个视图之后，在查询时可以完全将它作为一个table使用
sqlline> !connect jdbc:calcite:model=target/test-classes/model-with-view.json admin admin
sqlline> SELECT e.name, d.name FROM female_emps AS e JOIN depts AS d on e.deptno = d.deptno;
+--------+------------+
| NAME | NAME |
+--------+------------+
| Wilma | Marketing |
+--------+------------+ 自定义表是由用户定义的代码来实现定义的，不需要额外自定义schema。
具体例子请参考 model-with-custom-table.json
{
version: '1.0',
defaultSchema: 'CUSTOM_TABLE',
schemas: [{
name: 'CUSTOM_TABLE',
tables: [
{
name: 'EMPS',
type: 'custom',
factory: 'org.apache.calcite.adapter.csv.CsvTableFactory',
operand: {
file: 'target/test-classes/sales/EMPS.csv.gz',
flavor: "scannable"
}
}
]
}
]
}我们可以通过通用的方式来对自定义表进行查询
sqlline> !connect jdbc:calcite:model=target/test-classes/model-with-custom-table.json admin admin
sqlline> SELECT empno, name FROM custom_table.emps;
+--------+--------+
| EMPNO | NAME |
+--------+--------+
| 100 | Fred |
| 110 | Eric |
| 110 | John |
| 120 | Wilma |
| 130 | Alice |
+--------+--------+上面的schema是通用格式，包含了一个由org.apache.calcite.adapter.csv.CsvTableFactory驱动的自定义表，这个类实现了Calcite中的TableFactory接口。它创建了一个CsvScannableTable实例方法，将model文件中的file参数传递过去。
public CsvTable create(SchemaPlus schema, String name,
Map<String, Object> map, RelDataType rowType) {
String fileName = (String) map.get("file");
final File file = new File(fileName);
final RelProtoDataType protoRowType =
rowType != null ? RelDataTypeImpl.proto(rowType) : null;
return new CsvScannableTable(file, protoRowType);
} 实现自定义table通常是一个比实现自定义schema更容易的替代方法。这两种方法最终都会创建类似的Table接口类的实现，但自定义表无需实现metadata discovery。（CsvTableFactory创建一个CsvScannableTable，就像CsvSchema一样，但是table实现无需扫描整个文件系统来找到.csv类型的文件）。
自定义table要求开发者在model上执行更多操作（开发者需要在model文件中显式指定每一个table和它对应的文件），同时也提供给了开发者更多的控制选项（例如，为每一个table提供不同参数）。
model定义过程中可以通过/…/或者//符号来添加注释
{
version: '1.0',
/* Multi-line
comment. */
defaultSchema: 'CUSTOM_TABLE',
// Single-line comment.
schemas: [
..
]
}Comments不是标准JSON格式，但不会造成影响。
目前我们看到的table实现和查询都没有问题，因为table中不会包含大数据量。但如果自定义table数据量大，例如，一百列，100w行，你会希望用户在每次查询过程中不要检索全量数据。你会希望Calcite通过适配器来进行衡量，并找到一个更有效的方法来访问数据。
衡量过程是一个简单的查询优化格式。Calcite支持通过添加计划器规则来实现查询优化。计划器规则在查询解析树中匹配到对应规则时生效(例如在某个项目中匹配到某种类型的table时生效)，而且计划器规则是可扩展的，例如schemas和tables。因此，如果用户希望通过SQL访问某个数据集，首先需要定义一个自定义表或是schema，然后再去定义一些能使数据访问高效的规则。
为了查看效果，我们可以使用一个计划器规则来访问一个CSV文件中的某些子列集合。我们可以在两个相似的schema中执行同样的查询：
sqlline> !connect jdbc:calcite:model=target/test-classes/model.json admin admin
sqlline> explain plan for select name from emps;
+-----------------------------------------------------+
| PLAN |
+-----------------------------------------------------+
| EnumerableCalcRel(expr#0..9=[{inputs}], NAME=[$t1]) |
| EnumerableTableScan(table=[[SALES, EMPS]]) |
+-----------------------------------------------------+
sqlline> !connect jdbc:calcite:model=target/test-classes/smart.json admin admin
sqlline> explain plan for select name from emps;
+-----------------------------------------------------+
| PLAN |
+-----------------------------------------------------+
| EnumerableCalcRel(expr#0..9=[{inputs}], NAME=[$t1]) |
| CsvTableScan(table=[[SALES, EMPS]]) |
+-----------------------------------------------------+ 两次查询的scan方式不同，EnumerableTableScan和CsvTableScan。
是什么导致了执行计划的差异？让我们来追查一下其中证据。在smart.json model 文件中，存在额外的一行：
flavor: "translatable" 这个配置会让CsvSchema携带falvor = TRANSLATABLE 参数进行创建，并且它的createTable方法会创建CsvTranslatableTable实例，而不是CsvScannableTable.
CsvTranslatableTable实现了TranslatableTable.toRel()方法来创建CsvTableScan. Table scan操作是查询执行树中的叶子节点，默认实现方式是EnumerableTableScan，但我们构造了一种不同的的子类型来让规则生效。
下面是一个完整的规则：
public class CsvProjectTableScanRule extends RelOptRule {
public static final CsvProjectTableScanRule INSTANCE =
new CsvProjectTableScanRule();
private CsvProjectTableScanRule() {
super(
operand(Project.class,
operand(CsvTableScan.class, none())),
"CsvProjectTableScanRule");
}
@Override
public void onMatch(RelOptRuleCall call) {
final Project project = call.rel(0);
final CsvTableScan scan = call.rel(1);
int[] fields = getProjectFields(project.getProjects());
if (fields == null) {
// Project contains expressions more complex than just field references.
return;
}
call.transformTo(
new CsvTableScan(
scan.getCluster(),
scan.getTable(),
scan.csvTable,
fields));
}
private int[] getProjectFields(List<RexNode> exps) {
final int[] fields = new int[exps.size()];
for (int i = 0; i < exps.size(); i++) {
final RexNode exp = exps.get(i);
if (exp instanceof RexInputRef) {
fields[i] = ((RexInputRef) exp).getIndex();
} else {
return null; // not a simple projection
}
}
return fields;
}
} 构造函数声明了能使规则生效的关系表达式匹配模式。
onMatch方法生成了一个新的关系表达式，调用了RelOptRuleCall.transformTo()来表示规则已经触发成功了。
关于Calcite的查询计划有多智能有许多种方法，但我们在这里不会讨论这个问题。最智能的部分是为了减轻用户负担的优化器规则设计者。
首先，Calcite不会按照规定的顺序来执行规则。查询优化处理过程是一个有很多分支的分支树，就像国际象棋一样会检查很多可能的子操作（移动）。如果规则A和B同时满足查询操作树的一个给定子集合，Calcite可以将它们同时执行。
其次，Calcite在执行计划树的时候会使用基于代价的优化，但基于成本的模型并不会导致规则的执行，这在短期内看起来代价会更大。
许多优化规则都有一个线性优化方案。在面对上面说的规则A和规则B的情况下，这样的优化器需要立刻进行抉择。可能会有一个策略，比如“在整棵树上先执行规则A，然后在整棵树上执行规则B”，或是执行基于代价的优化策略，执行能产生耗费更低的结果的规则。
Calcite不需要这样的妥协（哪样？？？）。这能让结合各种规则的操作更简单。如果你希望结合规则来识别各种携带规则的物化视图，去从CSV和JDBC源数据系统中读取数据，需要给Calcite所有的规则并告诉它如何去做。
Calcite使用了一个基于成本的优化模型，成本模型决定了最终使用哪个执行计划，有时候为了避免搜索空间的爆炸性增长会对搜索树进行剪枝，但它绝不对强迫用户在规则A和规则B之间进行选择。这是很重要的一点，因为它避免了在搜索空间中落入实际上不是最优的局部最优值。
同样，成本模型是可插拔(扩展)的，它是基于表和查询操作的统计信息。这个问题稍后会仔细讨论。
JDBC适配器将JDBC数据源中的schema映射成了Calcite的schema模式。
例如，下面这个schema是从一个MySQL 的“foodmart”数据库中读取出来的。
{
version: '1.0',
defaultSchema: 'FOODMART',
schemas: [
{
name: 'FOODMART',
type: 'custom',
factory: 'org.apache.calcite.adapter.jdbc.JdbcSchema$Factory',
operand: {
jdbcDriver: 'com.mysql.jdbc.Driver',
jdbcUrl: 'jdbc:mysql://localhost/foodmart',
jdbcUser: 'foodmart',
jdbcPassword: 'foodmart'
}
}
]
}(The FoodMart database will be familiar to those of you who have used the Mondrian OLAP engine, because it is Mondrian’s main test data set. To load the data set, follow Mondrian’s installation instructions.)
当前的限制：JDBC适配器目前仅支持下推table scan操作；其他的的操作（filtering，joins，aggregations等等）在Calcite中完成。我们的目的是将尽可能多的处理操作、语法转换、数据类型和内建函数下推到源数据系统。如果一个Calcite查询来源于单独一个JDBC数据库中的表，从原则上来说整个查询都会下推到源数据系统中。如果表来源于多个JDBC数据源，或是一个JDBC和非JDBC的混合源，Calcite会使用尽可能高效的分布式查询方法来完成本次查询。
克隆JDBC适配器创造了一个混合源数据系统。数据来源于JDBC数据库但在它第一次读取时会读取到内存表中。Calcite基于内存表对查询进行评估，有效地实现了数据库的缓存。
例如，下面的model从MySQL数据库中读取“foodmart”表:
{
version: '1.0',
defaultSchema: 'FOODMART_CLONE',
schemas: [
{
name: 'FOODMART_CLONE',
type: 'custom',
factory: 'org.apache.calcite.adapter.clone.CloneSchema$Factory',
operand: {
jdbcDriver: 'com.mysql.jdbc.Driver',
jdbcUrl: 'jdbc:mysql://localhost/foodmart',
jdbcUser: 'foodmart',
jdbcPassword: 'foodmart'
}
}
]
}另外一种技术是从当前已存在的schema中构建一份clone schema。通过source属性来引用之前已经在model中定义过的schema，如下：
{
version: '1.0',
defaultSchema: 'FOODMART_CLONE',
schemas: [
{
name: 'FOODMART',
type: 'custom',
factory: 'org.apache.calcite.adapter.jdbc.JdbcSchema$Factory',
operand: {
jdbcDriver: 'com.mysql.jdbc.Driver',
jdbcUrl: 'jdbc:mysql://localhost/foodmart',
jdbcUser: 'foodmart',
jdbcPassword: 'foodmart'
}
},
{
name: 'FOODMART_CLONE',
type: 'custom',
factory: 'org.apache.calcite.adapter.clone.CloneSchema$Factory',
operand: {
source: 'FOODMART'
}
}
]
} 可以使用这种方法建立任意类型schema的clone schema，不仅限于JDBC.
cloning adapter不是最重要的。我们计划开发更复杂的缓存策略，和更复杂更有效的内存表的实现，但目前cloning JDBC adapter仅体现了这种可能性，并让我们能开始尝试初始实现。
There are many other ways to extend Calcite not yet described in this tutorial. The adapter specification describes the APIs involved.
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
