本文共 14467 字，大约阅读时间需要 48 分钟。

翻译原文：

1.引言

Calcite作为一个强大的SQL计算引擎，在Flink内部的SQL引擎模块就是基于Calcite。但是目前Calcite的相关学习文档还是太少了，没有系统的学习资料。

2. 教程

这是一个循序渐进的教程，演示如何建立和连接Calcite。它使用一个简单的适配器，使CSV文件的目录看起来像一个包含表的schema。Calcite完成剩下的工作，并提供完整的SQL接口。

Calcite-example-CSV是Calcite的完全功能适配器，用于读取CSV（逗号分隔值）格式的文本文件。值得注意的是，几百行Java代码足以提供完整的SQL查询能力。

CSV 还可以用作构建其他数据格式适配器的模板。尽管代码行不多，但它涵盖了几个重要概念：

• 使用SchemaFactory和模式接口的用户自定义Schema
• 在模型JSON文件中声明Schemas；
• 在模型JSON文件中声明视图
• 利用Table接口来自定义表
• 确定表的记录类型
• 表的简单实现，使用ScannableTable 接口，直接枚举所有行
• 一种更高级的实现，实现FilterableTable，并可以根据简单的谓词筛选出行
• 使用TranslatableTable，该表使用计划器规则转换为关系运算符

3. 下载和编译

您需要Java（版本8, 9或10）和Git：

$ git clone https://github.com/apache/calcite.git$ cd calcite$ ./mvnw install -DskipTests -Dcheckstyle.skip=true$ cd example/csv

4. 首次查询

现在让我们使用sqlline连接到Calcite，这是一个包含在这个项目中的sql shell。

$ ./sqllinesqlline> !connect jdbc:calcite:model=target/test-classes/model.json admin admin

（如果运行的是Windows，则命令为sqlline.bat）

执行元数据查询：

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元数据的命令，比如 !columns和!describe。)

如您所见，系统中有5个表：表EMPS, DEPTS和HOBBIES是在当前SALES schema当中； COLUMNS和TABLES是在系统元数据schema中。系统表始终存在于Calcite中，但其他表由schema的具体实现提供。在这种情况下，EMPS和DEPTS表基于target/test-classes目录中的EMPS.csv和DEPTS.csv文件。

让我们对这些表执行一些查询，以显示Calcite提供了SQL的完整实现。首先，表扫描：

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特征。我们没有时间在这里报道他们。再写一些查询进行实验。

5. Schema发现

Calcite是怎么找到这些tables的呢？记住，Calcite core对csv文件一无所知。作为“没有存储层的数据库”，Calcite不知道任何文件格式。Calcite知道这些表，因为我们告诉它在calcite-example-csv项目中运行代码。

在那条链上有几个步骤。首先，我们基于schema文件中的schema工厂类定义schema。然后schema工厂创建一个schema，schema创建几个表，每个表都知道如何通过扫描csv文件来获取数据。最后，在Calcite解析了查询并计划使用这些表之后，Calcite在执行查询时调用这些表来读取数据。现在让我们更详细地看看这些步骤。

在jdbc connect字符串中，我们给出了json格式的模型路径。模型如下：

{
     version: '1.0',  defaultSchema: 'SALES',  schemas: [    {
         name: 'SALES',      type: 'custom',      factory: 'org.apache.calcite.adapter.csv.CsvSchemaFactory',      operand: {
           directory: 'target/test-classes/sales'      }    }  ]}

该模型定义了一个名为“SALES”的schema。该schema由org.apache.calcite.adapter.csv.CsvSchemaFactory类提供支持，该插件类是calcite-example-csv项目的一部分，实现了Calcite 接口SchemaFactory。它的create方法实例化一个schema，从模型文件传入directory参数：

public Schema create(SchemaPlus parentSchema, String name,    Map
   
     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);}
   

在模型的驱动下，schema工厂实例化一个称为“SALES”的单一模式。该模式是org.apache.calcite.adapter.csv.CsvSchema的一个实例，实现了Calcite interface Schema。

schema的工作是生成表的列表。（它也可以列出子schema和表函数，但这些是高级功能，并且calcite-example-csv不支持它们。）这些表实现 Calcite的Table接口。CsvSchema生成的表是CsvTable及其子类的实例。

下面是CsvSchema的相关代码， 重写了AbstractSchema 基类的 getTableMap()方法。

protected Map
   
     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
    
      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扫描directory并且发现那些.csv文件，并且为它们创建表。在本案例中，directory是 target/test-classes/sales，然后包含了EMPS.csv和DEPTS.csv，这成为了表EMPS和DEPTS。

6. schemas中的表和视图

注意我们不需要在schema中定义任何表；schema会自动生成表。您可以使用schema的tables属性定义额外的表，而不是自动创建的表。让我们看看如何创建一个重要且有用的表类型，即视图。

在编写查询时，视图看起来像一个表，但它不存储数据。它通过执行查询来获得结果。在计划查询的同时展开视图，因此查询规划器通常可以执行优化，例如从select子句中删除最终结果中未使用的表达式。

下面是定义了视图的schema：

{
     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\''        }      ]    }  ]}

JSON不容易编写长字符串，因此Calcite支持另一种语法。如果视图有一个长的SQL语句，则可以提供行列表，而不是单个字符串。

{
     name: 'FEMALE_EMPS',  type: 'view',  sql: [    'SELECT * FROM emps',    'WHERE gender = \'F\''  ]}

现在我们定义了一个视图，我们可以在查询中使用它，就像它是一个表一样：

sqlline> SELECT e.name, d.name FROM female_emps AS e JOIN depts AS d on e.deptno = d.deptno;+--------+------------+|  NAME  |    NAME    |+--------+------------+| Wilma  | Marketing  |+--------+------------+

7. 自定义表

自定义表是由用户自定义的代码驱动其实现的表。它们不需要活在自定义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 adminsqlline> SELECT empno, name FROM custom_table.emps;+--------+--------+| EMPNO  |  NAME  |+--------+--------+| 100    | Fred   || 110    | Eric   || 110    | John   || 120    | Wilma  || 130    | Alice  |+--------+--------+

该schema是常规schema，包含由org.apache.calcite.adapter.csv.CsvTableFactory,提供支持的自定义表，该表实现了Calcite接口TableFactory。其create方法实例化一个CsvScannableTable，从模型文件传入文件参数：

public CsvTable create(SchemaPlus schema, String name,    Map
   
     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);}
   

实现自定义表通常是实现自定义schema的简单替代方案。这两种方法最终可能会创建一个类似的Table接口实现，但是对于自定义表，您不需要实现元数据发现。（CsvTableFactory与 CsvSchema一样创建CsvScannableTable，但表实现不会扫描文件系统中的.csv文件。）

自定义表需要为模型做更多的工作（作者需要显式地指定每个表及其文件），但也给作者更多的控制（例如，为每个表提供不同的参数）。

8. models里的注释

模型的注释可以用/* ... */ 和 //语义：

{
     version: '1.0',  /* Multi-line     comment. */  defaultSchema: 'CUSTOM_TABLE',  // Single-line comment.  schemas: [    ..  ]}

（注释不是标准JSON，但是是无害的扩展。）

9. 使用计划器规则优化查询

到目前为止，只要表不包含大量数据，我们看到的表实现就可以了。但是，如果您的自定义表有一百列和一百万行，那么您希望系统不会检索每个查询的所有数据。您希望Calcite与适配器协商，并找到一种更有效的访问数据的方法。

此协商是一种简单的查询优化形式。Calcite通过添加计划规则支持查询优化。规划器规则通过在查询解析树中查找模式（例如某个表顶部的项目）来操作，并将树中匹配的节点替换为一组新的节点来实现优化。

规划器规则也可以扩展，比如schemas和表。因此，如果您有一个要通过SQL访问的数据存储，那么首先要定义一个自定义表或schema，然后定义一些规则来提高访问效率。

要在实际操作中看到这一点，让我们使用计划器规则从csv文件访问列的子集。让我们针对两个非常相似的schemas运行相同的查询：

sqlline> !connect jdbc:calcite:model=target/test-classes/model.json admin adminsqlline> 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 adminsqlline> explain plan for select name from emps;+-----------------------------------------------------+| PLAN                                                |+-----------------------------------------------------+| EnumerableCalcRel(expr#0..9=[{inputs}], NAME=[$t1]) ||   CsvTableScan(table=[[SALES, EMPS]])               |+-----------------------------------------------------+

是什么导致了计划的不同？让我们跟踪证据的线索。在smart.json模型文件中，只有一行：

flavor: "translatable"

这将导致使用flavor = TRANSLATABLE来创建CsvSchema ，其createtable方法创建 CsvTranslatableTable实例，而不是CsvScannableTable实例。

CsvTranslatableTable实现了TranslatableTable.toRel()方法来创建CsvTableScan。表扫描是查询运算符树的叶。通常的实现是 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
   
     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()以指示规则已成功激发。

10. 查询优化过程

关于Calcite的查询计划有多聪明，有很多话要说，但这里我们不说。

首先，Calcite不会按规定的顺序触发规则。查询优化过程遵循分支树的许多分支，就像下棋程序检查许多可能的移动序列一样。如果规则A和B都匹配查询运算符树的给定部分，则Calcite可以同时触发这两个部分。

第二，Calcite使用成本模型来选择计划，但成本模型并不能阻止规则在短期内触发似乎更昂贵的计划。

许多优化器都有一个线性优化方案。面对规则A和规则B之间的选择，如上所述，这样的优化器需要立即进行选择。它可能有一个策略，例如“将规则A应用于整棵树，然后将规则B应用于整棵树”，或者应用基于成本的策略，应用产生更便宜结果的规则。

Calcite不需要这样的妥协。这使得组合各种规则集变得简单。如果，假设您想结合规则来识别物化视图（materialized views ）和从CSV及JDBC源系统中读取的规则，你只要给Calcite一套规则，让它去做。

Calcite确实使用成本模型。成本模型决定最终使用哪一个计划，有时会修剪搜索树以防止搜索空间爆炸，但它从不强制您在规则A和规则B之间进行选择。这一点很重要，因为它可以避免陷入搜索空间中实际不是最佳的局部极小值。

另外（您猜对了），成本模型是可插拔的，它所基于的表和查询操作符统计也是可插拔的。但这可能是以后的主题。

11. JDBC适配器

JDBC适配器将JDBC数据源中的schema映射为Calcite 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'      }    }  ]}

当前限制：JDBC适配器当前只向下推表扫描操作；所有其他处理（过滤、联接、聚合等）都发生在Calcite中。我们的目标是将尽可能多的处理推送到源系统，并在执行过程中转换语法、数据类型和内置函数。如果Calcite查询基于单个JDBC数据库中的表，那么原则上，整个查询应该转到该数据库。如果表来自多个JDBC源，或者是JDBC和非JDBC的混合，Calcite将使用最有效的分布式查询方法。

12. JDBC适配器的克隆

克隆JDBC适配器创建一个混合数据库。数据来源于JDBC数据库，但在第一次访问每个表时，数据被读取到内存中的表中。Calcite根据内存表中的查询来评估查询，这实际上是数据库的一个缓存。

例如，以下模型从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的基础上构建克隆schema。使用source属性引用模型中前面定义的架构，如下所示：

{
     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'      }    }  ]}

您可以使用这种方法在任何类型的模式上创建克隆模式，而不仅仅是JDBC。

克隆适配器不是万能的。我们计划开发更复杂的缓存策略，以及更完整和高效的内存表实现，但目前，克隆JDBC适配器展示了什么是可能的，并允许我们尝试最初的实现。

转载地址：https://blog.csdn.net/hxcaifly/article/details/88661252 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！