Apache Calcite 文档翻译 - 基础教程

tyrantlucifer

发布于 2022-03-23 14:53:58

9500

发布于 2022-03-23 14:53:58

文章被收录于专栏：Tyrant Lucifer

教程

这是一个手把手并循序渐进的教程，展示了如何和Calcite建立连接。它使用了一个简单的适配器，使得一个包含了csv文件的目录看起来是一个包含数据库表的模式（schema）。Calcite负责其他工作，并提供了一个完整的SQL接口。

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

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

使用SchemaFactory和Schema接口实现用户自定义的模式
在一个模型的JSON文件声明了模式
在一个模型的JSON文件声明了视图
使用Table接口实现用户自定义的表
确定了一个表的记录类型
一个表的最简单实现，实现了ScannableTable接口，直接枚举所有的行
一个更高级的实现，实现了FilterableTable接口，使得用户可以根据简单的谓词过滤出行
表的高级实现，实现了TranslatableTable接口，使其可以使用规划器规则翻译成关系运算符

下载构建

你至少需要安装Java 8/9/10和git工具

$ git clone https://github.com/apache/calcite.git
$ cd calcite/example/csv
$ ./sqlline

第一次查询

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

$ ./sqlline
sqlline> !connect jdbc:calcite:model=src/test/resources/model.json admin admin

（如果你是运行在Windows下，那么命令应该是.\sqlline.bat）

执行这条元数据查询语句：

sqlline> !tables
+-----------+-------------+------------+--------------+---------+----------+------------+-----------+---------------------------+----------------+
| TABLE_CAT | TABLE_SCHEM | TABLE_NAME |  TABLE_TYPE  | REMARKS | TYPE_CAT | TYPE_SCHEM | TYPE_NAME | SELF_REFERENCING_COL_NAME | REF_GENERATION |
+-----------+-------------+------------+--------------+---------+----------+------------+-----------+---------------------------+----------------+
|           | SALES       | DEPTS      | TABLE        |         |          |            |           |                           |                |
|           | SALES       | EMPS       | TABLE        |         |          |            |           |                           |                |
|           | SALES       | SDEPTS     | TABLE        |         |          |            |           |                           |                |
|           | metadata    | COLUMNS    | SYSTEM TABLE |         |          |            |           |                           |                |
|           | metadata    | TABLES     | SYSTEM TABLE |         |          |            |           |                           |                |
+-----------+-------------+------------+--------------+---------+----------+------------+-----------+---------------------------+----------------+

（JDBC专家请注意，!tables只是在背后执行了[DatabaseMetaData.getTables()](https://docs.oracle.com/javase/7/docs/api/java/sql/DatabaseMetaData.html#getTables(java.lang.String, java.lang.String, java.lang.String, java.lang.String[]))。还有其他命令可以查询JDBC元数据，比如!columns和!describe）

如你所见，在系统中有5张表：当前SALES模式中的表EMPS、DEPTS和SDEPTS，以及系统元数据(metadata)模式中的COLUMNS和TABLES，系统表在Calcite中一直存在，但其他表是由模式的具体实现提供的；在这种情况下，EMPS、DEPTS、SDEPTS表是基于resource/sale目录下的EMPS.csv.gz、DEPTS.csv、SDEPTS.csv文件所映射出来的。

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

sqlline> SELECT * FROM emps;
+-------+-------+--------+--------+---------------+-------+------+---------+---------+------------+
| EMPNO | NAME  | DEPTNO | GENDER |     CITY      | EMPID | AGE  | SLACKER | MANAGER |  JOINEDAT  |
+-------+-------+--------+--------+---------------+-------+------+---------+---------+------------+
| 100   | Fred  | 10     |        |               | 30    | 25   | true    | false   | 1996-08-03 |
| 110   | Eric  | 20     | M      | San Francisco | 3     | 80   |         | false   | 2001-01-01 |
| 110   | John  | 40     | M      | Vancouver     | 2     | null | false   | true    | 2002-05-03 |
| 120   | Wilma | 20     | F      |               | 1     | 5    |         | true    | 2005-09-07 |
| 130   | Alice | 40     | F      | Vancouver     | 2     | null | false   | true    | 2007-01-01 |
+-------+-------+--------+--------+---------------+-------+------+---------+---------+------------+

接下来执行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是如何发现这些表的？请牢记一点，Calcite的核心对CSV文件一无所知（作为一个“没有存储层的数据库”，Calcite不知道任何文件格式）。Calcite之所以知道这些表，是因为我们告诉他要运行calcite-example-csv项目中的代码。

在这个模式发现过程中，有好几个步骤。首先，我们根据模型文件中的模式工厂类来定义一个模式，然后模式工厂创建了一个模式，模式创建了几个表，每个表都知道如何通过扫描csv文件获得数据。最后，在Calcite解析了查询并计划使用这些表后，Calcite在执行查询时调用这些表来读取数据，现在让我们更详细地了解一下这些步骤。

在JDBC连接字符串中，我们给出了一个JSON格式的模型的路径。以下是该模型：

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

该模型定义了一个单一模式SALES，该模式由一个插件类org.apache.calcite.adapter.csv.CsvSchemaFactory提供，它是calcite-example-csv项目的一部分，实现了Calcite接口SchemaFactory。它的create方法实例化了一个模式，从模型文件中解析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);
}

在模式的驱动之下，模式工厂实例化了一个名为SALES的单一模式。这个模式是类org.apache.calcite.adapter.csv.CsvSchema的一个实例，实现了Calcite的接口Schema。

模式的工作是去产生一系列的表（它还可以列出子主题和表格功能，但这些都是高级功能，calite-example-csv并不支持它们），这些表实现了Calcite的Table接口。CsvSchema产生的表是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);
  }
}

模式扫描该目录，找到所有具有适当扩展名的文件，并为它们创建表。在本例中，目录是sales，包含文件EMPS.csv.gz、DEPTS.csv和SDEPTS.csv，这些文件成为表EMPS、DEPTS和SDEPTS。

模式中的表和视图

请注意，我们不需要在模型中定义任何表；模式自动生成了这些表。

你可以使用模式的表属性定义额外的表，而不是那些自动创建的表。

让我们看看如何创建一个重要而有用的表类型，即视图：

{
  version: '1.0',
  defaultSchema: 'SALES',
  schemas: [
    {
      name: 'SALES',
      type: 'custom',
      factory: 'org.apache.calcite.adapter.csv.CsvSchemaFactory',
      operand: {
        directory: 'sales'
      },
      tables: [
        {
          name: 'FEMALE_EMPS',
          type: 'view',
          sql: 'SELECT * FROM emps WHERE gender = \'F\''
        }
      ]
    }
  ]
}

type属性标识了FEMALE_EMPS是一个视图，而不是一个普通的表或者是自定义表。请注意，视图定义中的单引号是用反斜杠转义的，与JSON的正常方式相同。

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  |
+--------+------------+

自定义表

自定义表是由用户定义的代码驱动实现的表，所以他们并不需要在自定义模式中。

这个一个自定义表的例子存在于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: 'sales/EMPS.csv.gz',
            flavor: "scannable"
          }
        }
      ]
    }
  ]
}

我们可以使用通常的方式查询该表：

sqlline> !connect jdbc:calcite:model=src/test/resources/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  |
+--------+--------+

这个模式是一个常规模式，包含由类org.apache.calcite.adapter.csv.CsvTableFactory驱动的自定义表，该表实现了Calcite的接口TableFactory。它的创建方法实例化了一个CsvScannableTable，从模型文件中解析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接口的实现，但是对于自定义表，你不需要实现元数据发现（CsvTableFactory创建了一个CsvScannableTable，就像CsvSchema一样，但是表的实现并不扫描文件系统中的.csv文件）。

在模式定义中添加注释

模式JSON中可以使用/**/或//为配置项添加注释：

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

（注释不是标准的JSON，而是一种无害的扩展）

使用规则器规则优化查询

到目前为止，我们所看到的表的实现，只要表不包含大量的数据，就没有问题。但是，如果你的客户表有一百个列和一百万行，你希望系统不要为每次查询都检索所有的数据。你希望Calcite与适配器协商，找到一种更有效的方法来访问数据。

这种协商是查询优化的一种简单形式。Calcite通过添加规划器规则支持查询优化。规划器规则的操作方式是寻找查询解析树中的模式（例如类表上的项目），并通过一组实现优化的新节点来替换树中的匹配节点。

规划器规则也是可扩展的，就像模式和表一样。因此，如果你有一个想通过SQL访问的数据存储，你首先需要定义一个自定义的模式或表，然后再定义一些规则，使查询更有效率。

为了验证这一点，让我使用一个规划器规则来访问CSV文件中的一个列子集。让我们针对两个非常相似的模式运行相同的查询：

sqlline> !connect jdbc:calcite:model=src/test/resources/model.json admin admin
sqlline> explain plan for select name from emps;
+-----------------------------------------------------+
| PLAN                                                |
+-----------------------------------------------------+
| EnumerableCalc(expr#0..9=[{inputs}], NAME=[$t1])    |
|   EnumerableTableScan(table=[[SALES, EMPS]])        |
+-----------------------------------------------------+
sqlline> !connect jdbc:calcite:model=src/test/resources/smart.json admin admin
sqlline> explain plan for select name from emps;
+-----------------------------------------------------+
| PLAN                                                |
+-----------------------------------------------------+
| CsvTableScan(table=[[SALES, EMPS]], fields=[[1]])   |
+-----------------------------------------------------+

是什么导致了两个不同的查询计划？让我们跟随证据的线索，在smart.json模型文件中，只有一个额外的配置行：

flavor: "translatable"

这个配置项导致了CsvSchema创建了一个CsvTranslateTable表实例而不是一个CsvScannableTable实例

CsvTranslatableTable实现了TranslatableTable.toRel()方法来创建CsvTableScan。表扫描是查询运算操作树的叶子。通常的实现是EnumerableTableScan，但我们已经创建了一个独特的子类型，它将导致规则的启动。

以下是该规则的全部内容：

public class CsvProjectTableScanRule
    extends RelRule<CsvProjectTableScanRule.Config> {
  /** Creates a CsvProjectTableScanRule. */
  protected CsvProjectTableScanRule(Config config) {
    super(config);
  }

  @Override public void onMatch(RelOptRuleCall call) {
    final LogicalProject 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;
  }

  /** Rule configuration. */
  public interface Config extends RelRule.Config {
    Config DEFAULT = EMPTY
        .withOperandSupplier(b0 ->
            b0.operand(LogicalProject.class).oneInput(b1 ->
                b1.operand(CsvTableScan.class).noInputs()))
        .as(Config.class);

    @Override default CsvProjectTableScanRule toRule() {
      return new CsvProjectTableScanRule(this);
    }
}

该规则的默认实例驻留在CsvRules类中：

public abstract class CsvRules {
  public static final CsvProjectTableScanRule PROJECT_SCAN =
      CsvProjectTableScanRule.Config.DEFAULT.toRule();
}

对默认配置中withOperandSupplier方法的调用（接口Config中的DEFAULT字段）声明了将导致规则启动的关系表达式模式。如果规则器看到一个LogicalProject的唯一输入是一个没有输入的CsvTableScan，它将调用该规则。

onMatch方法生成一个新的关系表达式，并调用RelOptRuleCall.transformTo()来表示规则已经成功启动。

查询优化过程

关于Calcite的查询计划器有多聪明和强大，有很多话可以聊，但我们不会在这里说。聪明的设计是为了减轻你这个规划器规则编写者的负担。

首先，Calcite并不按规定的顺序执行规则。查询优化过程遵循分支树的许多分支，就像国际象棋游戏程序检查许多可能的动作序列。如果规则A和B都与查询运算树的某个部分相匹配，那么Calcite可以同时启动这两个规则。

第二，Calcite在选择计划时使用了成本模型，但成本模型并不妨碍抛弃那些在短期内执行会更昂贵的规则。

许多优化器有一个线性优化方案。如上所述，面对规则A和规则B的选择，这样的优化器需要立即做出选择。它可能有一个政策，如 "对整个树应用规则A，然后对整个树应用规则B"，或者应用一个基于成本的策略，应用产生更优结果的规则

但Calcite并不需要这样的妥协，这使得组合各种规则会变得更加简单一些，如果你想把识别物化视图的规则与读取CSV和JDBC源系统的规则结合起来，你只需给Calcite所有的规则集并告诉它去做就够了。

Calcite确实使用了成本模型，成本模型最终会决定最终使用哪个计划，有时也会修剪搜索树以防止搜索空间爆炸，但它不会强迫你在规则A和规则之间做出选择。这一点很重要，因为它可以避免陷入搜索空间中的局部最小值，而实际上这并不是最优解。

另外（你猜对了），成本模型是可插拔的，就像它所基于的表和查询操作符的统计一样。但这可以作为以后的一个主题再去细讲。

JDBC适配器

JDBC适配器将JDBC数据源中的模式映射为Calcite模式。

例如，这个模式读取了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'
      }
    }
  ]
}

(FoodMart数据库对于那些使用过Mondrian OLAP引擎的人来说会很熟悉，因为它是Mondrian的主要测试数据集。要加载这个数据集，请遵循Mondrian的安装说明）。

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

克隆的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'
      }
    }
  ]
}

另一种技术是在现有模式的基础上建立一个克隆模式。你使用源属性来引用模型中早期定义的模式，像这样：

{
  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 适配器展示了可能的情况，并允许我们尝试我们的初始实现。