11-15 2,681 views
什么是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个步骤:
- 将SQL解析成抽象语法树;
- 对抽象语法树进行校验;
- 将抽象语法树解析成关系代数表达式;
- 对关系代数表达式进行优化,在保持语义不变的前提下,转化为较优的表达式;
- 将优化后关系代数表达式转化为物理执行计划并计划,返回最终的结果。
目前Calcite在大数据和数据存储领域有着广泛的使用,如表所示:
Kylin、Phoenix、Hive、Flink等均使用Calcite实现JDBC驱动、SQL解析、SQL校验、SQL 查询优化等。
Kylin查询源码分析
数据模型
在源码分析时,我们使用Kylin的官方数据模型示例,如图所示:
该雪花模型包含以下5张表:
- KYLIN_SALES,销售事实表;
- KYLIN_ACCOUNT,用户维度表;
- KYLIN_CAL_DT,日期维度表;
- KYLIN_CATEGORY_GROUPING,类别维度表
- KYLIN_COUNTR,国家维度表。
示例SQL如下所示:
1 |
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方法,如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 |
@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相关接口实例,部分代码如下:
1 2 3 4 5 6 7 8 9 10 |
@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方法,该方法的核心代码如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
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方法中,如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
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方法创建,代码如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
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字段传入元数据描述文件
元数据
元数据描述文件示例如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 |
{ "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集合,代码如下所示:
1 2 3 |
public Map<String, Table> getTableMap() { return buildTableMap(); } |
OLAPTable类实现了Calcite的QueryableTable和TranslatableTable接口(这两个接口均继承自Table接口),用于描述表,其中比较重要的几个方法如下所示:
1 2 3 4 5 6 7 8 |
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扫描数据。
1 2 3 4 |
public Statistic getStatistic() { List<ImmutableBitSet> keys = new ArrayList<ImmutableBitSet>(); return Statistics.of(100, keys); } |
该方法在Table接口中定义,用于获取表的行数等统计信息,在CBO方式的优化中计算成本,但是Kylin存储引擎统计的是各Cuboid的统计信息,所以这里统一返回固定值。
1 2 3 4 5 6 |
@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类),如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
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中,如下所示:
1 2 3 4 5 6 7 8 9 |
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方法进行校验,校验的细节此处暂不展开,仅列出校验的主要步骤包括:
- 对抽象语法树根节点进行标准化重写,在保持语义的前提下,将SqlOrderBy、SqlDelete、SqlUpdate、SqlMerge等类型节点重写为SqlSelect类型节点;
- 调用各节点的validate方法进行校验。
代码如下:
1 2 3 4 5 6 7 8 |
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方法进一步解析成关系代数表达式。这个方法代码如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
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类型节点的代码部分如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 |
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等子节点进行转化。最终,上一步的抽象语法树进行转化后得到下列关系代数表达式:
1 2 3 4 5 6 7 |
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类:
- 基于规则的优化器(Rule Based Optimizer,简称RBO):根据优化规则将一个关系表达式转化为另外一个关系表达式,同时原有表达式被放弃,经过一系列转化后生成最终表达式;
- 基于成本的优化器(Cost Based Optimizer,简称CBO):根据优化规则对关系表达式进行转换,同时原有表达式也会保留,经过一系列转化后生成多个表达式,之后计算每个表达式的成本,从中挑选成本最小的表达式作为最终表达式。
Calcite有两个优化器实现:
- HepPlanner: RBO的实现,按照规则进行匹配,直到达到次数限制或者遍历后无匹配规则;
- VolcanoPlanner: CBO 的实现,一直迭代各规则,直到找到成本最小的表达式。
Calcite的优化在Prepare类的optimize方法中,部分代码如下所示:
1 2 3 4 5 6 7 |
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类实例,代码如下所示:
1 2 3 4 5 6 7 8 |
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扩展的多个规则,如下表所示:
类 | 说明 |
---|---|
OLAPToEnumerableConverterRule | 将RelNode类型节点转化为OLAPToEnumerableConverter类型节点 |
OLAPFilterRule | 将LogicalFilter类型节点转化为OLAPFilterRel类型节点 |
OLAPProjectRule | 将LogicalProject类型节点转化为OLAPProjectRel类型节点 |
OLAPAggregateRule | 将LogicalAggregate类型节点转化为OLAPAggregateRel类型节点 |
OLAPJoinRule | 将LogicalJoin 类型节点转化为 OLAPJoinRel或OLAPFilterRel类型节点 |
OLAPLimitRule | 将Sort类型节点转化为OLAPLimitRel类型节点 |
OLAPSortRule | 将Sort类型节点转化为OLAPSortRel类型节点 |
OLAPUnionRule | 将Union类型节点转化为OLAPUnionRel类型节点 |
OLAPWindowRule | 将Window类型节点转化为OLAPWindowRel类型节点 |
OLAPValuesRule | 将LogicalValues类型节点转化为OLAPValuesRel类型节点 |
同时,还删除了多个Calcite原生的规则,部分如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
// 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); |
以此来保持表达式中的一些模式,便于后续物理执行计划的相关操作。经过上述优化后,关系表达式转化为如下结果:
1 2 3 4 5 6 7 8 9 |
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]])<code> |
执行
最后是将关系代数表达式转化为实际物理执行计划,扫描HBase中的数据并返回结果,这里正在梳理过程中,可先参考官方推荐的文档https://www.jianshu.com/p/21df8303d2ae了解其原理,其中,会执行OLAPTableScan类的implement方法,如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
@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方法,代码如下所示:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
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数据。
版权属于: 我爱我家
原文地址: http://magicwt.com/2020/11/15/kylin%e6%9f%a5%e8%af%a2%e6%ba%90%e7%a0%81%e5%88%86%e6%9e%90/
转载时必须以链接形式注明原始出处及本声明。