仓库源文站点原文


layout: post title: "Apache Calcite 框架 50 倍性能优化实践" categories: Calcite tags: 性能优化 Calcite Apache

author: 张乘辉

初步认识 Calcite

项目使用 Calcite 框架作为查询引擎,之前一直没停过这玩意是啥,而且网上资料特别少,又是体现我学习能力的时候了,在着手排查性能问题前,我花了非常多时间在了解 Calcite 实现原理上面。

1、Calcite 简介

Apache Calcite是一款开源的动态数据管理框架,它提供了标准的 SQL 语言、多种查询优化和连接各种数据源的能力,但不包括数据存储、处理数据的算法和存储元数据的存储库。

Calcite 之前的名称叫做optiq,optiq 起初在 Hive 项目中,为 Hive 提供基于成本模型的优化,即CBO(Cost Based Optimizatio)。2014 年 5 月 optiq 独立出来,成为 Apache 社区的孵化项目,2014 年 9 月正式更名为 Calcite。

Calcite 的目标是“one size fits all(一种方案适应所有需求场景)”,希望能为不同计算平台和数据源提供统一的查询引擎。

2、Calcite 执行流程

1)解析 SQL,目的是为了将 SQL 转换成 AST 抽象语法数,Calcite 有一个专门的对象 SqlNode 表示;

2)语法检查,用数据库的元数据信息进行语法验证;

3)逻辑优化,根据前面生成的逻辑计划按照相应的规则(Rule)进行优化;

4)SQL 执行,按照执行计划执行。

3、Calcite 相关对象

关系表达式, 主要有 TableScan, Project, Sort, Join 等。如果 SQL 为查询的话,所有关系达式都可以在 SqlSelect中找到, 如 where 和 having 对应的 Filter, selectList 对应 Project, orderBy、offset、fetch 对应着 Sort, From 对应着 TableScan/Join 等等, 示便 Sql 最后会生成如下 RelNode 树。

Debug 源码得到的 RelNode 对象长这样:

行表达式, 如 RexLiteral(常量), RexCall(函数), RexInputRef (输入引用) 等,举个例子:

SELECT LOCATION as LOCATION,MERGE2(VALUE2) as VALUE2 
FROM transaction 
WHERE REPORTTIME >=1594887720000 AND REPORTTIME <=1594891320000 AND APPID = 'test-api'  AND GROUP2 IN ('DubboService','URL') AND METRICKEY IN ('$$TOTAL') GROUP BY LOCATION

RexCall

<=($1, 1595496539000)

RexInputRef

$1

RexLiteral

1595496539000:BIGINT

下面根据官方资料的描述,总结 Calcite 的三种查询模式:

1)ScannableTable

这种方式基本不会用,原因是查询数据库的时候没有任何条件限制,默认会先把全部数据拉到内存,然后再根据filter条件在内存中过滤。

使用方式:实现Enumerable scan(DataContext root);,该函数返回Enumerable对象,通过该对象可以一行行的获取这个Table的全部数据。

2)FilterableTable

初级用法,我们能拿到filter条件,即能再查询底层DB时进行一部分的数据过滤,一般开始介入calcite可以用这种方式(translatable方式学习成本较高)。

使用方式:实现Enumerable scan(DataContext root, List filters )

如果当前类型的“表”能够支持我们自己写代码优化这个过滤器,那么执行完自定义优化器,可以把该过滤条件从集合中移除,否则,就让calcite来过滤,简言之就是,如果我们不处理List filters ,Calcite也会根据自己的规则在内存中过滤,无非就是对于查询引擎来说查的数据多了,但如果我们可以写查询引擎支持的过滤器(比如写一些hbase、es的filter),这样在查的时候引擎本身就能先过滤掉多余数据,更加优化。提示,即使走了我们的查询过滤条件,可以再让calcite帮我们过滤一次,比较灵活。

3)TranslatableTable

高阶用法,有些查询用上面的方式都支持不了或支持的不好,比如join、聚合、或对于select的字段筛选等,需要用这种方式来支持,好处是可以支持更全的功能,代价是所有的解析都要自己写,“承上启下”,上面解析sql的各个部件,下面要根据不同的DB(esmysqldrudi..)来写不同的语法查询。

当使用ScannableTable的时候,我们只需要实现函数Enumerable scan(DataContext root);,该函数返回Enumerable对象,通过该对象可以一行行的获取这个Table的全部数据(也就意味着每次的查询都是扫描这个表的数据,我们干涉不了任何执行过程);当使用FilterableTable的时候,我们需要实现函数Enumerable scan(DataContext root, List filters );参数中多了filters数组,这个数据包含了针对这个表的过滤条件,这样我们根据过滤条件只返回过滤之后的行,减少上层进行其它运算的数据集;当使用TranslatableTable的时候,我们需要实现RelNode toRel( RelOptTable.ToRelContext context, RelOptTable relOptTable);,该函数可以让我们根据上下文自己定义表扫描的物理执行计划,至于为什么不在返回一个Enumerable对象了,因为上面两种其实使用的是默认的执行计划,转换成EnumerableTableAccessRel算子,通过TranslatableTable我们可以实现自定义的算子,以及执行一些其他的rule,Kylin就是使用这个类型的Table实现查询。

由于我对 Calcite 还没有一个更加深入的了解(但是并不阻碍我排查问题,而且 Calcite 这玩意对我来说太复杂了),因此 Calcite 更加复杂的概念我在这里就不继续啰嗦了,比如关系代数的基本知识、查询优化器等等,排查问题并不需要了解那么深入,而且项目中只是使用了 Calcite 一小部分功能。

使用 Calcite 实现一个简单的数据库

需要做如下几步:

  1. 编写 model.json
  2. 自定义 SchemaFactory
  3. 自定义 Schema(像一个“没有存储层的databse”一样,Calcite不会去了解任何文件格式)
  4. 自定义Table
  5. 自定义 Enumerator

demo url: https://github.com/objcoding/calcite-demo

耗时排查

我在项目中使用了 FilterableTable 模式,Cacite 在解析 Sql 时耗时非常大,然后通过调试,我发现每个请求都占据了两个位置:

org.apache.calcite.adapter.enumerable.EnumerableInterpretable#getBindable

Cacite 在这个地方通过设置缓存大小来优化缓存设置。

org.apache.calcite.interpreter.JaninoRexCompiler#baz

但是不会缓存该位置,并且每次都会使用新的表达式字符串通过反射创建它。

我使用 JProfile 工具对线程耗时的地方进行定位:

Calcite 会在这个地方会调用反射根据不同的 Sql 动态生成不同的表达式,Debug 获取的表达式如下:

Calcite 为什么会有这种机制呢?我们先从 Bindable 对象讲起:

在 EnumerableRel(RelNode,我们可以通过 TranslatableTable 自定义 FilterRel、JoinRel、AggregateRel)的每个算子的 implement 方法中会将一些算子(Group、join、sort、function)要实现的算法写成 Linq4j 的表达式,然后通过这些 Linq4j 表达式生成 Java Class。通过 JavaRowFormat 格式化)

calcite 会将 sql 生成的 linq4j 表达式生成可执行的 Java 代码( Bindable 类): org.apache.calcite.adapter.enumerable.EnumerableInterpretable#getBindable

Calcite 会调用 Janino 编译器动态编译这个 java 类,并且实例化这个类的一个对象,然后将其封装到 CalciteSignature 对象中。

调用 executorQuery 查询方法并创建 CalciteResultSet 的时候会调用 Bindable 对象的 bind 方法,这个方法返回一个Eumerable对象:

org.apache.calcite.avatica.AvaticaResultSet#execute

org.apache.calcite.jdbc.CalcitePrepare.CalciteSignature#enumerable

将 Enumerable 赋值给 CalciteResultSet 的 cursor 成员变量。

在执行真正的数据库查询时,获得实际的 CalciteResultSet,最终会调用:

org.apache.calcite.avatica.AvaticaResultSet#next

以下是根据 SQL 动态生成的 linq4j 表达式:

public static class Record2_0 implements java.io.Serializable {
  public Object f0;
  public boolean f1;
  public Record2_0() {}
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (!(o instanceof Record2_0)) {
      return false;
    }
    return java.util.Objects.equals(this.f0, ((Record2_0) o).f0) && this.f1 == ((Record2_0) o).f1;
  }

  public int hashCode() {
    int h = 0;
    h = org.apache.calcite.runtime.Utilities.hash(h, this.f0);
    h = org.apache.calcite.runtime.Utilities.hash(h, this.f1);
    return h;
  }

  public int compareTo(Record2_0 that) {
    int c;
    c = org.apache.calcite.runtime.Utilities.compare(this.f1, that.f1);
    if (c != 0) {
      return c;
    }
    return 0;
  }

  public String toString() {
    return "{f0=" + this.f0 + ", f1=" + this.f1 + "}";
  }

}

public org.apache.calcite.linq4j.Enumerable bind(final org.apache.calcite.DataContext root) {
  final org.apache.calcite.rel.RelNode v1stashed = (org.apache.calcite.rel.RelNode) root.get("v1stashed");
  final org.apache.calcite.interpreter.Interpreter interpreter = new org.apache.calcite.interpreter.Interpreter(
    root,
    v1stashed);
  java.util.List accumulatorAdders = new java.util.LinkedList();
  accumulatorAdders.add(new org.apache.calcite.linq4j.function.Function2() {
    public Record2_0 apply(Record2_0 acc, Object[] in) {
      final Object inp9_ = in[9];
      if (inp9_ != null) {
        acc.f1 = true;
        acc.f0 = com.zto.zcat.store.api.query.Merge2Fun.add(acc.f0, inp9_);
      }
      return acc;
    }
    public Record2_0 apply(Object acc, Object in) {
      return apply(
        (Record2_0) acc,
        (Object[]) in);
    }
  }
  );
  org.apache.calcite.adapter.enumerable.AggregateLambdaFactory lambdaFactory = new org.apache.calcite.adapter.enumerable.BasicAggregateLambdaFactory(
    new org.apache.calcite.linq4j.function.Function0() {
      public Object apply() {
        Object a0s0;
        boolean a0s1;
        a0s1 = false;
        a0s0 = com.zto.zcat.store.api.query.Merge2Fun.init();
        Record2_0 record0;
        record0 = new Record2_0();
        record0.f0 = a0s0;
        record0.f1 = a0s1;
        return record0;
      }
    }
,
    accumulatorAdders);
  return org.apache.calcite.linq4j.Linq4j.singletonEnumerable(interpreter.aggregate(lambdaFactory.accumulatorInitializer().apply(), lambdaFactory.accumulatorAdder(), lambdaFactory.singleGroupResultSelector(new org.apache.calcite.linq4j.function.Function1() {
      public Object apply(Record2_0 acc) {
        return acc.f1 ? com.zto.zcat.store.api.query.Merge2Fun.result(acc.f0) : (Object) null;
      }
      public Object apply(Object acc) {
        return apply(
          (Record2_0) acc);
      }
    }
    )));
}


public Class getElementType() {
  return java.lang.Object.class;
}

Enumerator是Linq风格的迭代器,它有4个方法:

  1. current
  2. moveNext
  3. reset
  4. close

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

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

总结执行顺序:

1、executeQuery 方法:

1)根据算子 linq4j 表达式子生成 Bindable 执行对象,如果有设置缓存,则会将对像存储到缓存中;

2)生成 CalciteResultSet 时会调用 Bindable#bind 方法返回一个 Enumerable 对象;

2、getData 方法:调用 ResultSet#next 方法最终会嗲用 Enumerable#moveNext

一图理解 Bindable 在 calcite 中的作用:

发现 Bindable 缓存会持续增加,说明 Bindable 类内容不一致:

也说明了 calcite 会根据不同的 SQL 动态生成 linq4j 表达式。

性能优化

以上排查结果可知,在 Calcite 内容会频繁使用 JaninoRexCompiler 使用反射动态生成表达式,由于项目中的 sql 格式相对固定,因此我们是否可以自定义一个 Compiler,然后替换 JaninoRexCompiler ?

我将使用 JaninoRexCompiler 的相关类复制出来,实现一个自定义的 Interpreter.ScalarCompiler,然后在这个地方 org.apache.calcite.interpreter.Interpreter.CompilerImpl#CompilerImpl替换 JaninoRexCompiler。

关于自定义 Interpreter.ScalarCompiler 的具体思路过程,我记录在这里了:

https://issues.apache.org/jira/browse/CALCITE-4144

经过反复调试,发现性能提上了 50 倍以上!

再次使用 JProfiler 查看,发现 Calcite 查询过程耗时已经大大降低了。