什么是Kylin

Apache Kylin是一个开源的、分布式的分析型数据仓库，提供Hadoop/Spark 之上的 SQL 查询接口及多维分析（OLAP）能力以支持超大规模数据，最初由 eBay 开发并贡献至开源社区。它能在亚秒内查询巨大的表。

Kylin的查询高性能主要依赖于Cube理论，如图所示：

它将表字段划分为维度和量度，通过预先计算，在维度上进行量度聚合并保存聚合结果，而根据维度进行聚合查询时，则可以命中已保存的聚合结果，大大减少数据扫描量和实时计算量。

Kylin的系统架构如图所示：

它依赖大数据基础设施HBase、Spark、Hadoop等实现分布式的存储和计算，并基于这些基础设施，设计了构建引擎和查询引擎来分别实现数据的构建和查询。在查询引擎部分，Kylin使用了Calcite来实现SQL的解析、优化和执行。

什么是Calcite

Calcite是一个用于优化异构数据源的查询处理的基础框架，提供了标准的 SQL 语言、多种查询优化和连接各种数据源的能力。从功能上看，它支持SQL 解析、SQL 校验、SQL 查询优化、SQL 生成、数据连接查询等，但不包括数据处理和存储。Calcite的架构如图所示：

数据处理和存储系统提供元数据和规则至Calcite，Calcite提供JDBC Server面向客户端查询，并对SQL查询请求进行处理，在Calcite中，一个SQL的解析和执行大概经过以下5个步骤：

1. 将SQL解析成抽象语法树；
2. 对抽象语法树进行校验；
3. 将抽象语法树解析成关系代数表达式；
4. 对关系代数表达式进行优化，在保持语义不变的前提下，转化为较优的表达式；
5. 将优化后关系代数表达式转化为物理执行计划并计划，返回最终的结果。

目前Calcite在大数据和数据存储领域有着广泛的使用，如表所示：

Kylin、Phoenix、Hive、Flink等均使用Calcite实现JDBC驱动、SQL解析、SQL校验、SQL 查询优化等。

Kylin查询源码分析

数据模型

在源码分析时，我们使用Kylin的官方数据模型示例，如图所示：

该雪花模型包含以下5张表：

1. KYLIN_SALES，销售事实表；
2. KYLIN_ACCOUNT，用户维度表；
3. KYLIN_CAL_DT，日期维度表；
4. KYLIN_CATEGORY_GROUPING，类别维度表
5. KYLIN_COUNTR，国家维度表。

示例SQL如下所示：

select s.lstg_site_id,sum(s.price) as price_sum from kylin_sales as s inner join kylin_account as a on s.buyer_id = a.account_id where a.account_country='US' group by s.lstg_site_id order by price_sum desc limit 10

用于查询各站点来自美国消费者的销售额。

入口

Kylin支持多种查询入口，包括WEB控制台、REST API、JDBC驱动、ODBC驱动等。这里介绍了一下JDBC驱动的实现。

当客户端使用JDBC接口访问时，加载的JDBC驱动是org.apache.kylin.jdbc.Driver，这里，Kylin使用了JDBC驱动框架Avatica，Driver即继承自org.apache.calcite.avatica.UnregisteredDriver，覆盖了getFactoryClassName方法，如下所示：

@Override
    protected String getFactoryClassName(JdbcVersion jdbcVersion) {
        switch (jdbcVersion) {
        case JDBC_30:
            throw new UnsupportedOperationException();
        case JDBC_40:
            return KylinJdbcFactory.Version40.class.getName();
        case JDBC_41:
        default:
            return KylinJdbcFactory.Version41.class.getName();
        }
}

该段代码说明仅支持JDBC 4.0即以上版本协议，并返回了相应的工厂类。工厂类KylinJdbcFactory实现了AvaticaFactory接口，用于创建JDBC相关接口实例，部分代码如下：

@Override
public AvaticaStatement newStatement(AvaticaConnection connection, StatementHandle h, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
    return new KylinStatement((KylinConnection) connection, h, resultSetType, resultSetConcurrency, resultSetHoldability);
}

@Override
public AvaticaResultSet newResultSet(AvaticaStatement statement, QueryState state, Signature signature, TimeZone timeZone, Frame firstFrame) throws SQLException {
    AvaticaResultSetMetaData resultSetMetaData = new AvaticaResultSetMetaData(statement, null, signature);
    return new KylinResultSet(statement, state, signature, resultSetMetaData, timeZone, firstFrame);
}

该段代码说明JDBC相关接口的实现是在Avatica框架实现基础的进一步继承和扩展，例如，ResultSet接口实现是KylinResultSet，其是AvaticaResultSet的子类，KylinResultSet主要覆盖了AvaticaResultSet的execute方法，该方法的核心代码如下所示：

KylinConnection connection = (KylinConnection) statement.connection;
IRemoteClient client = connection.getRemoteClient();

Map<String, String> queryToggles = new HashMap<>();
int maxRows = statement.getMaxRows();
queryToggles.put("ATTR_STATEMENT_MAX_ROWS", String.valueOf(maxRows));
addServerProps(queryToggles, connection);

QueryResult result;
try {
    result = client.executeQuery(sql, paramValues, queryToggles);
} catch (IOException e) {
    throw new SQLException(e);
}

该段代码说明获取IRemoteClient实例，执行executeQuery方法返回查询结果，而从IRemoteClient的实现KylinClient中的代码可以看到，其实质上是调用kylin-server模块的REST API “kylin/api/query”来实现查询。

那具体看一下该API的服务端，略过Controller层、缓存命中、注释删除等环节，查询代码在org.apache.kylin.rest.service.QueryService的executeRequest方法中，如下所示：

Pair<List<List<String>>, List<SelectedColumnMeta>> r = null;
try {
    stat = conn.createStatement();
    processStatementAttr(stat, sqlRequest);
    resultSet = stat.executeQuery(correctedSql);

    r = createResponseFromResultSet(resultSet);

} catch (SQLException sqlException) {
    r = pushDownQuery(sqlRequest, correctedSql, conn, sqlException);
    if (r == null)
        throw sqlException;

    isPushDown = true;
} finally {
    close(resultSet, stat, null); //conn is passed in, not my duty to close
}

该段代码仍是基于JDBC接口，而connection通过org.apache.kylin.query.QueryConnection的getConnection方法创建，代码如下所示：

if (!isRegister) {
    try {
        Class<?> aClass = Thread.currentThread().getContextClassLoader()
                .loadClass("org.apache.calcite.jdbc.Driver");
        Driver o = (Driver) aClass.getDeclaredConstructor().newInstance();
        DriverManager.registerDriver(o);
    } catch (ClassNotFoundException | InstantiationException | IllegalAccessException | NoSuchMethodException | InvocationTargetException e) {
        e.printStackTrace();
    }
    isRegister = true;
}
File olapTmp = OLAPSchemaFactory.createTempOLAPJson(project, KylinConfig.getInstanceFromEnv());
Properties info = new Properties();
info.putAll(KylinConfig.getInstanceFromEnv().getCalciteExtrasProperties());
// Import calcite props from jdbc client(override the kylin.properties)
info.putAll(BackdoorToggles.getJdbcDriverClientCalciteProps());
info.put("model", olapTmp.getAbsolutePath());
info.put("typeSystem", "org.apache.kylin.query.calcite.KylinRelDataTypeSystem");
return DriverManager.getConnection("jdbc:calcite:", info);

其加载的JDBC驱动是org.apache.calcite.jdbc.Driver，说明后续基于Calcite进行SQL解析、校验、优化和执行。这里，通过Calcite标准的model字段传入元数据描述文件

元数据

元数据描述文件示例如下所示：

{
    "version": "1.0",
    "defaultSchema": "DEFAULT",
    "schemas": [
        {
            "type": "custom",
            "name": "DEFAULT",
            "factory": "org.apache.kylin.query.schema.OLAPSchemaFactory",
            "operand": {
                "project": "learn_kylin"
            },
            "functions": [
               {
                   name: 'PERCENTILE',
                   className: 'org.apache.kylin.measure.percentile.PercentileAggFunc'
               },
               {
                   name: 'CONCAT',
                   className: 'org.apache.kylin.query.udf.ConcatUDF'
               },
               {
                   name: 'MASSIN',
                   className: 'org.apache.kylin.query.udf.MassInUDF'
               },
               {
                   name: 'INTERSECT_COUNT',
                   className: 'org.apache.kylin.measure.bitmap.BitmapIntersectDistinctCountAggFunc'
               },
               {
                   name: 'VERSION',
                   className: 'org.apache.kylin.query.udf.VersionUDF'
               },
               {
                   name: 'PERCENTILE_APPROX',
                   className: 'org.apache.kylin.measure.percentile.PercentileAggFunc'
               }
            ]
        }
    ]
}

OLAPSchemaFactory类实现了Calcite的SchemaFactory接口，创建OLAPSchema类实例，而OLAPSchema类则通过读取HBase上存储的元数据信息生成OLAPTable类的Map集合，代码如下所示：

public Map<String, Table> getTableMap() {
    return buildTableMap();
}

OLAPTable类实现了Calcite的QueryableTable和TranslatableTable接口（这两个接口均继承自Table接口），用于描述表，其中比较重要的几个方法如下所示：

public RelDataType getRowType(RelDataTypeFactory typeFactory) {
    if (this.rowType == null) {
        // always build exposedColumns and rowType together
        this.sourceColumns = getSourceColumns();
        this.rowType = deriveRowType(typeFactory);
    }
    return this.rowType;
}

该方法在Table接口中定义，用于获取表字段信息，Kylin除了按Calcite定义返回RelDataType类型的字段信息，也会按自身定义保存ColumnDesc类型的字段集合，其中包含维度信息，用于后续执行物理查询时判断从哪些Cuboid扫描数据。

public Statistic getStatistic() {
    List<ImmutableBitSet> keys = new ArrayList<ImmutableBitSet>();
    return Statistics.of(100, keys);
}

该方法在Table接口中定义，用于获取表的行数等统计信息，在CBO方式的优化中计算成本，但是Kylin存储引擎统计的是各Cuboid的统计信息，所以这里统一返回固定值。

@Override
public RelNode toRel(ToRelContext context, RelOptTable relOptTable) {
    int fieldCount = relOptTable.getRowType().getFieldCount();
    int[] fields = identityList(fieldCount);
    return new OLAPTableScan(context.getCluster(), relOptTable, this, fields);
}

该方法在TranslatableTable接口中定义，说明在优化过程中，扫描表数据的关系表达式节点会转化为OLAPTableScan类实例，关于OLAPTableScan类会在后续优化、执行过程中再介绍。

同时，OLAPTable还有一些方法用于返回Enumerable接口实例（也就是Kylin实现的OLAPQuery类），如下所示：

public Enumerable<Object[]> executeOLAPQuery(DataContext optiqContext, int ctxSeq) {
    return new OLAPQuery(optiqContext, EnumeratorTypeEnum.OLAP, ctxSeq);
}

public Enumerable<Object[]> executeLookupTableQuery(DataContext optiqContext, int ctxSeq) {
    return new OLAPQuery(optiqContext, EnumeratorTypeEnum.LOOKUP_TABLE, ctxSeq);
}

public Enumerable<Object[]> executeColumnDictionaryQuery(DataContext optiqContext, int ctxSeq) {
    return new OLAPQuery(optiqContext, EnumeratorTypeEnum.COL_DICT, ctxSeq);
}

public Enumerable<Object[]> executeHiveQuery(DataContext optiqContext, int ctxSeq) {
    return new OLAPQuery(optiqContext, EnumeratorTypeEnum.HIVE, ctxSeq);
}

这些方法在执行物理查询时，Calcite会通过反射进行调用，获取OLAPQuery类，实际的数据扫描委托给OLAPQuery完成，关于OLAPQuery类会在后续优化、执行过程中再介绍。

解析

查询请求传递到服务端，并在服务端执行stat.executeQuery(correctedSql)后，会进入具体的查询流程，首先进行SQL的解析。Calcite的语法解析是基于JavaCC实现的。JavaCC是一个语法解析器生成框架， 其根据预先定义的规则生成相应的解析器代码。Calcite的语法解析规则是calcite-core中的codegen/templates/Parser.jj，根据其生成的解析器类是org.apache.calcite.sql.parser.impl. SqlParserImpl。调用该类实例解析SQL生成抽象语法树（即SqlNode）的代码在CalcitePrepareImpl中，如下所示：

SqlParser parser = createParser(query.sql,  parserConfig);
SqlNode sqlNode;
try {
  sqlNode = parser.parseStmt();
  statementType = getStatementType(sqlNode.getKind());
} catch (SqlParseException e) {
  throw new RuntimeException(
      "parse failed: " + e.getMessage(), e);
}

示例SQL经过解析得到的抽象语法树如图所示：

树中的每个节点是SqlNode子类实例。

校验

抽象语法树通过SqlValidatorImpl类的validate方法进行校验，校验的细节此处暂不展开，仅列出校验的主要步骤包括：

1. 对抽象语法树根节点进行标准化重写，在保持语义的前提下，将SqlOrderBy、SqlDelete、SqlUpdate、SqlMerge等类型节点重写为SqlSelect类型节点；
2. 调用各节点的validate方法进行校验。

代码如下：

SqlNode outermostNode = performUnconditionalRewrites(topNode, false);
cursorSet.add(outermostNode);
top = outermostNode;
TRACER.trace("After unconditional rewrite: {}", outermostNode);
if (outermostNode.isA(SqlKind.TOP_LEVEL)) {
  registerQuery(scope, null, outermostNode, outermostNode, null, false);
}
outermostNode.validate(this, scope);

所以，上一步的抽象语法树经过校验会转化为下图：

关系代数表达式

校验后的抽象语法树通过SqlToRelConverter 类的convertQueryRecursive方法进一步解析成关系代数表达式。这个方法代码如下所示：

final SqlKind kind = query.getKind();
switch (kind) {
case SELECT:
  return RelRoot.of(convertSelect((SqlSelect) query, top), kind);
case INSERT:
  return RelRoot.of(convertInsert((SqlInsert) query), kind);
case DELETE:
  return RelRoot.of(convertDelete((SqlDelete) query), kind);
case UPDATE:
  return RelRoot.of(convertUpdate((SqlUpdate) query), kind);
case MERGE:
  return RelRoot.of(convertMerge((SqlMerge) query), kind);
case UNION:
case INTERSECT:
case EXCEPT:
  return RelRoot.of(convertSetOp((SqlCall) query), kind);
case WITH:
  return convertWith((SqlWith) query, top);
case VALUES:
  return RelRoot.of(convertValues((SqlCall) query, targetRowType), kind);
default:
  throw new AssertionError("not a query: " + query);

从这个方法的命名就可以看出，方法内部是在递归调用，从抽象语法树根节点开始，根据其类型，分别调用对应的convertSelect、convertInsert、convertDelete、convertUpdate、convertMerge等方法，转化为RelNode类型实例（RelNode是关系代数表达式节点接口），而在这些方法内部，会对传入节点的子节点，根据其类型，再递归调用这些方法，从而将抽象语法树所有节点均转化为RelNode类型实例，并建立相互之间的关系。通过convertSelect方法具体转化SqlSelect类型节点的代码部分如下所示：

convertFrom(
    bb,
    select.getFrom());
convertWhere(
    bb,
    select.getWhere());

final List<SqlNode> orderExprList = new ArrayList<>();
final List<RelFieldCollation> collationList = new ArrayList<>();
gatherOrderExprs(
    bb,
    select,
    select.getOrderList(),
    orderExprList,
    collationList);
final RelCollation collation =
    cluster.traitSet().canonize(RelCollations.of(collationList));

if (validator.isAggregate(select)) {
  convertAgg(
      bb,
      select,
      orderExprList);
} else {
  convertSelectList(
      bb,
      select,
      orderExprList);
}

if (select.isDistinct()) {
  distinctify(bb, true);
}

convertOrder(
    select, bb, collation, orderExprList, select.getOffset(),
    select.getFetch());

其依次对from、where、groupBy、orderBy等子节点进行转化。最终，上一步的抽象语法树进行转化后得到下列关系代数表达式：

LogicalSort(sort0=[$1], dir0=[DESC], fetch=[10])
  LogicalAggregate(group=[{0}], PRICE_SUM=[SUM($1)])
    LogicalProject(LSTG_SITE_ID=[$4], PRICE=[$5])
      LogicalFilter(condition=[=($16, 'US')])
        LogicalJoin(condition=[=($7, $13)], joinType=[inner])
          OLAPTableScan(table=[[DEFAULT, KYLIN_SALES]], ctx=[], fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]])
          OLAPTableScan(table=[[DEFAULT, KYLIN_ACCOUNT]], ctx=[], fields=[[0, 1, 2, 3, 4, 5]])

其根节点排序（LogicalSort），叶子节点是两张表的数据扫描（OLAPTableScan）

优化

在转化得到关系代数表达式后，会进一步对其进行优化，在保持语义不变的前提下，对表达式进行转化和调整以找到最优的表达式。优化器可以分为2类：

1. 基于规则的优化器（Rule Based Optimizer，简称RBO）：根据优化规则将一个关系表达式转化为另外一个关系表达式，同时原有表达式被放弃，经过一系列转化后生成最终表达式；
2. 基于成本的优化器（Cost Based Optimizer，简称CBO）：根据优化规则对关系表达式进行转换，同时原有表达式也会保留，经过一系列转化后生成多个表达式，之后计算每个表达式的成本，从中挑选成本最小的表达式作为最终表达式。

Calcite有两个优化器实现：

1. HepPlanner： RBO的实现，按照规则进行匹配，直到达到次数限制或者遍历后无匹配规则；
2. VolcanoPlanner： CBO 的实现，一直迭代各规则，直到找到成本最小的表达式。

Calcite的优化在Prepare类的optimize方法中，部分代码如下所示：

final RelOptPlanner planner = root.rel.getCluster().getPlanner();

…

final Program program = getProgram();
final RelNode rootRel4 = program.run(
    planner, root.rel, desiredTraits, materializationList, latticeList);

首先生成一个VolcanoPlanner类型的优化器实例，然后通过getProgram获取Program接口实例，并通过该接口的run方法执行优化。Program接口有多个实现，getProgram方法最终是通过Programs类的standard方法获取的SequenceProgram类实例，代码如下所示：

return sequence(subQuery(metadataProvider),
    new DecorrelateProgram(),
    new TrimFieldsProgram(),
    program1,

    // Second planner pass to do physical "tweaks". This the first time
    // that EnumerableCalcRel is introduced.
    calc(metadataProvider));

也就是说，优化可划分为串行执行的5步，包括将子查询转化为Join操作、删除无用字段等，其中第三步是使用已创建的VolcanoPlanner类型优化器实例进行优化。关系代数表达式的叶子节点是OLAPTableScan类型，该类覆盖了RelNode的register方法，其中添加了Kylin扩展的多个规则，如下表所示：

同时，还删除了多个Calcite原生的规则，部分如下所示：

// since join is the entry point, we can't push filter past join
planner.removeRule(FilterJoinRule.FILTER_ON_JOIN);
planner.removeRule(FilterJoinRule.JOIN);

// since we don't have statistic of table, the optimization of join is too cost
planner.removeRule(JoinCommuteRule.INSTANCE);
planner.removeRule(JoinPushThroughJoinRule.LEFT);
planner.removeRule(JoinPushThroughJoinRule.RIGHT);

// keep tree structure like filter -> aggregation -> project -> join/table scan, implementOLAP() rely on this tree pattern
planner.removeRule(AggregateJoinTransposeRule.INSTANCE);
planner.removeRule(AggregateProjectMergeRule.INSTANCE);
planner.removeRule(FilterProjectTransposeRule.INSTANCE);
planner.removeRule(SortJoinTransposeRule.INSTANCE);
planner.removeRule(JoinPushExpressionsRule.INSTANCE);
planner.removeRule(SortUnionTransposeRule.INSTANCE);
planner.removeRule(JoinUnionTransposeRule.LEFT_UNION);
planner.removeRule(JoinUnionTransposeRule.RIGHT_UNION);
planner.removeRule(AggregateUnionTransposeRule.INSTANCE);
planner.removeRule(DateRangeRules.FILTER_INSTANCE);
planner.removeRule(SemiJoinRule.JOIN);
planner.removeRule(SemiJoinRule.PROJECT);

以此来保持表达式中的一些模式，便于后续物理执行计划的相关操作。经过上述优化后，关系表达式转化为如下结果：

OLAPLimitRel(ctx=[], fetch=[10])
    OLAPSortRel(sort0=[$1], dir0=[DESC], ctx=[])
      OLAPAggregateRel(group=[{0}], PRICE_SUM=[SUM($1)], ctx=[])
        OLAPProjectRel(LSTG_SITE_ID=[$4], PRICE=[$5], ctx=[])
          OLAPFilterRel(condition=[=($16, 'US')], ctx=[])
            OLAPJoinRel(condition=[=($7, $13)], joinType=[inner], ctx=[])
              OLAPTableScan(table=[[DEFAULT, KYLIN_SALES]], ctx=[], fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]])
              OLAPTableScan(table=[[DEFAULT, KYLIN_ACCOUNT]], ctx=[], fields=[[0, 1, 2, 3, 4, 5]])

执行

最后是将关系代数表达式转化为实际物理执行计划，扫描HBase中的数据并返回结果，这里正在梳理过程中，可先参考官方推荐的文档https://www.jianshu.com/p/21df8303d2ae了解其原理，其中，会执行OLAPTableScan类的implement方法，如下所示：

@Override
public Result implement(EnumerableRelImplementor implementor, Prefer pref) {

    context.setReturnTupleInfo(rowType, columnRowType);
    String execFunction = genExecFunc();

    PhysType physType = PhysTypeImpl.of(implementor.getTypeFactory(), getRowType(), JavaRowFormat.ARRAY);
    MethodCallExpression exprCall = Expressions.call(table.getExpression(OLAPTable.class), execFunction,
            implementor.getRootExpression(), Expressions.constant(context.id));
    return implementor.result(physType, Blocks.toBlock(exprCall));
}

public String genExecFunc() {
    // if the table to scan is not the fact table of cube, then it's a lookup table
    if (context.realization.getModel().isLookupTable(tableName)) {
        return "executeLookupTableQuery";
    } else if (DictionaryEnumerator.ifDictionaryEnumeratorEligible(context)) {
        return "executeColumnDictionaryQuery";
    } else {
        return "executeOLAPQuery";
    }

}

通过反射调用OLAPQuery类的相关方法，这些方法在之前介绍元数据时曾提及，其均返回Enumerable接口实例。以OLAPEnumerator为例，其在迭代获取结果时，实质上是调用queryStorage方法，代码如下所示：

private ITupleIterator queryStorage() {

    logger.debug("query storage...");
    // bind dynamic variables
    olapContext.bindVariable(optiqContext);

    // If olapContext is cached, then inherit it.
    if (!olapContext.isBorrowedContext) {
        olapContext.resetSQLDigest();
    }
    SQLDigest sqlDigest = olapContext.getSQLDigest();

    // query storage engine
    IStorageQuery storageEngine = StorageFactory.createQuery(olapContext.realization);
    ITupleIterator iterator = storageEngine.search(olapContext.storageContext, sqlDigest,
            olapContext.returnTupleInfo);
    if (logger.isDebugEnabled()) {
        logger.debug("return TupleIterator...");
    }

    return iterator;
}

获取相应的IStorageQuery接口实例，通过协处理器扫描HBase数据。