以调试形式,深入理解凤凰全局索引

阅读数:642 2019 年 9 月 29 日 08:00

以调试形式,深入理解凤凰全局索引

第六章

Debug 之前,需要写一段简单代码,用以驱动调试过程。

复制代码
PhoenixConnection connection = (PhoenixConnection)DriverManager.getConnection(
"jdbc:phoenix:local:2181",
"test",
"test");
PhoenixStatement statement = (PhoenixStatement)connection.createStatement();
String sql = "select /*+ INDEX(test idx_test_email) */ * from test where email='test@test.com'";
PhoenixResultSet explainRes = (PhoenixResultSet)statement.executeQuery("explain "+sql);
while (explainRes.next()){
logger.info("explain: {}",explainRes.getString(1));
}
PhoenixResultSet resultSet = (PhoenixResultSet)statement.executeQuery(sql);
while (resultSet.next()){
logger.info("id: {},name: {},email: {}",
resultSet.getString("id"),
resultSet.getString("name"),
resultSet.getString("email"));
}
explainRes.close();
resultSet.close();
statement.close();
connection.close();
} catch (SQLException e) {
e.printStackTrace();
}

请注意上面的代码中:

connection/statement/resultSet 等类型已经被强制转换成了 Phoenix 对应的类型,这样方便 IDEA 直接跳转到具体的实现,否则就会跳转到 java 的 interface 中。这也算一个小技巧吧,调试其他源码的时候,最好转换成具体实现的类。

复制代码
CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN FULL SCAN OVER TEST
SKIP-SCAN-JOIN TABLE 0
CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN RANGE SCAN OVER IDX_TEST_EMAIL ['test@test.com']
SERVER FILTER BY FIRST KEY ONLY
DYNAMIC SERVER FILTER BY "TEST.ID" IN ($2.$4)

上面是 explain 得到的 SQL 执行计划,很明显 Phoenix 使用了 idx_test_email 这个索引。

根据之前文章的分析,我们可以在 executeQuery 内部设置一个断点,看一下解析后的 statement 的数据。

以调试形式,深入理解凤凰全局索引

SelectStatement 这个类的变量不算多,但涵盖了 select 查询的所有信息。此处我们只关注几个重要的变量:fromTable、hint、where。不过这里要特别注意 hint 变量,是一个 HintNode 类型,扩展全文索引的时候,会修改这个类的。

之前也分析过,SelectStatement 会被编译成 QueryPlan,下面是 QueryPlan 的具体字段值。

复制代码
plan = {ScanPlan@3907}
splits = null
scans = null
allowPageFilter = true
isSerial = false
isDataToScanWithinThreshold = false
serialRowsEstimate = null
serialBytesEstimate = null
serialEstimateInfoTs = null
tableRef = {TableRef@3910}
tableRefs = {SingletonImmutableSet@3911} size = 1
context = {StatementContext@3912}
resolver = {FromCompiler$ProjectedTableColumnResolver@3920}
binds = {BindManager@3921}
scan = {Scan@3922} "{"loadColumnFamiliesOnDemand":true,"filter":"EMAIL = 'test@test.com'","startRow":"","stopRow":"","batch":-1,"cacheBlocks":true,"totalColumns":1,"maxResultSize":-1,"families":{"0":["ALL"]},"caching":2147483647,"maxVersions":1,"timeRange":[0,9223372036854775807]}"
expressions = {ExpressionManager@3923}
aggregates = {AggregationManager@3924}
dateFormat = "yyyy-MM-dd HH:mm:ss.SSS"
dateFormatter = {FastDateFormat@3926} "FastDateFormat[yyyy-MM-dd HH:mm:ss.SSS]"
timeFormat = "yyyy-MM-dd HH:mm:ss.SSS"
timeFormatter = {FastDateFormat@3926} "FastDateFormat[yyyy-MM-dd HH:mm:ss.SSS]"
timestampFormat = "yyyy-MM-dd HH:mm:ss.SSS"
timestampFormatter = {FastDateFormat@3926} "FastDateFormat[yyyy-MM-dd HH:mm:ss.SSS]"
dateFormatTimeZone = {ZoneInfo@3927} "sun.util.calendar.ZoneInfo[id="GMT",offset=0,dstSavings=0,useDaylight=false,transitions=0,lastRule=null]"
numberFormat = "#,##0.###"
tempPtr = {ImmutableBytesWritable@3929} ""
statement = {PhoenixStatement@3903}
dataColumns = {LinkedHashMap@3930} size = 0
currentTime = -1
scanRanges = {ScanRanges@3931} "ScanRanges[[]]"
sequences = {SequenceManager@3932}
currentTable = {TableRef@3910}
whereConditionColumns = {ArrayList@3933} size = 1
subqueryResults = {HashMap@3934} size = 0
readMetricsQueue = {ReadMetricQueue@3935}
overAllQueryMetrics = {OverAllQueryMetrics@3936}
queryLogger = null
isClientSideUpsertSelect = false
statement = {PhoenixStatement$ExecutableSelectStatement@3904} "SELECT /*+ INDEX(TEST IDX_TEST_EMAIL) */ * FROM TEST WHERE EMAIL = 'test@test.com'"
projection = {RowProjector@3913} "[ID,NAME,EMAIL]"
paramMetaData = {PhoenixParameterMetaData@3914}
limit = null
offset = null
orderBy = {OrderByCompiler$OrderBy@3915}
groupBy = {GroupByCompiler$GroupBy$1@3916}
parallelIteratorFactory = {ParallelIteratorFactory$1@3917}
dynamicFilter = null
dataPlan = null
estimatedRows = null
estimatedSize = null
estimateInfoTimestamp = null
getEstimatesCalled = false

大家可以重点查看 scan 字段的逻辑,简单来看就是对 TEST 表的全文检索,filter 是 EMAIL=‘test@test.com’。到这里还没有走索引,需要继续执行代码。

复制代码
plan = connection.getQueryServices().getOptimizer().optimize(PhoenixStatement.this, plan);

上面代码是对 QueryPlan 进行优化,也就是执行计划优化。这里要具体看一下优化后的 plan 是怎么样的。

复制代码
plan = {HashJoinPlan@3967}
statement = {SelectStatement@3970} "SELECT /*+ NO_INDEX */ TEST.* FROM TEST Semi JOIN (SELECT /*+ INDEX(TEST IDX_TEST_EMAIL) */ 1 $3,":ID" $4 FROM "IDX_TEST_EMAIL" WHERE "0:EMAIL" = 'test@test.com') ON ("ID" = $2.$4)"
joinInfo = {HashJoinInfo@3971}
subPlans = {HashJoinPlan$HashSubPlan[1]@3972}
0 = {HashJoinPlan$HashSubPlan@3978}
index = 0
plan = {TupleProjectionPlan@3979}
tupleProjector = {TupleProjector@3984} "TUPLE-PROJECTOR {[1, "ID"] ==> [INTEGER, VARCHAR]}"
postFilter = null
delegate = {ScanPlan@3985}
splits = null
scans = null
allowPageFilter = true
isSerial = false
isDataToScanWithinThreshold = false
serialRowsEstimate = null
serialBytesEstimate = null
serialEstimateInfoTs = null
tableRef = {TableRef@3987}
tableRefs = {SingletonImmutableSet@3988} size = 1
context = {StatementContext@3989}
statement = {SelectStatement@3990} "SELECT /*+ INDEX(TEST IDX_TEST_EMAIL) */ 1 $3,":ID" $4 FROM "IDX_TEST_EMAIL" WHERE "0:EMAIL" = 'test@test.com'"
projection = {RowProjector@3991} "[1,"ID"]"
paramMetaData = {PhoenixParameterMetaData@3992}
limit = null
offset = null
orderBy = {OrderByCompiler$OrderBy@3993}
groupBy = {GroupByCompiler$GroupBy$1@3994}
parallelIteratorFactory = {ParallelIteratorFactory$1@3995}
dynamicFilter = null
dataPlan = null
estimatedRows = null
estimatedSize = null
estimateInfoTimestamp = null
getEstimatesCalled = false
hashExpressions = null
singleValueOnly = false
keyRangeLhsExpression = {RowKeyColumnExpression@3980} ""TEST.ID""
keyRangeRhsExpression = {ProjectedColumnExpression@3981} "$2.$4"
recompileWhereClause = false
tableRefs = {HashSet@3974} size = 2
maxServerCacheTimeToLive = 30000
serverCacheLimit = 104857600
dependencies = {HashMap@3975} size = 0
hashClient = null
firstJobEndTime = null
keyRangeExpressions = null
estimatedRows = null
estimatedBytes = null
estimateInfoTs = null
getEstimatesCalled = false
delegate = {ScanPlan@3976}

这是优化后的执行计划,SQL 语句被优化成 semi join 了。有没有很熟悉的感觉?作者在 HBase Meetup 中,有提到过这种优化方案,就是直接用 join 强制走索引。那么这里有必要简单介绍一下 semi joiin 的概念。

所谓的 semi-join 是指 semi-join 子查询。 当一张表在另一张表找到匹配的记录之后,半连接 ( semi-jion ) 返回第一张表中的记录。与条件连接相反,即使在右节点中找到几条匹配的记录,左节点 的表也只会返回一条记录。另外,右节点的表一条记录也不会返回。半连接通常使用 IN 或 EXISTS 作为连接条件。
https://blog.csdn.net/lppl010_/article/details/80301757

作者竟误打误撞,竟然猜出了 Phoenix 索引使用的原理。读者一定要深刻理解这个机制,有助于我们实现全文检索索引。也要注意优化后的 plan 类型是 HashJoinPlan。

复制代码
/**
*
* Interface for an executable query plan
*
*
* @since 0.1
*/
public interface QueryPlan extends StatementPlan ;
/**
* Get a result iterator to iterate over the results
* @return result iterator for iterating over the results
* @throws SQLException
*/
public ResultIterator iterator() throws SQLException;
/**
*
* @return whether underlying {@link ResultScanner} can be picked up in a round-robin
* fashion. Generally, selecting scanners in such a fashion is possible if rows don't
* have to be returned back in a certain order.
* @throws SQLException
*/
public boolean useRoundRobinIterator() throws SQLException;

在分析 HashJoinPlan 之前,先看一下 QueryPlan 接口的定义及其重要的函数。

iterator 返回一个 ResultIterator 用来迭代数据。那么 HashJoinPlan 的这个 iterator 方法就非常重要了,当然 iterator 有几种不同的重载形式,也需要关注下。

根据对 HashJoinPlan 的调试追踪,定位到了以下函数:

复制代码
public ResultIterator iterator(ParallelScanGrouper scanGrouper, Scan scan) throws SQLException

以调试形式,深入理解凤凰全局索引

这个函数的代码有点多,但仔细分析其逻辑,可以知道该函数前半段,将涉及到的子查询放到 Future 后台执行,所有子查询结束后其数据放到 ServerCache,dependencies 存放各个子查询对应的 ServerCache。最终用 dependencies 改写当前 QueryPlan 的 delegate。简单来说就是后台并行执行子查询,将结果用以改写当前执行计划的 scan 对象。

复制代码
iterator = {RoundRobinResultIterator@4319}
threshold = 2147483646
numScannersCacheExhausted = 0
resultIterators = {ParallelIterators@4325} "ResultIterators [name=PARALLEL,id=8f373aba-e360-94d0-ae67-3774e06c338e,scans=[[{"loadColumnFamiliesOnDemand":true,"startRow":"1","stopRow":"1\\x00","batch":-1,"cacheBlocks":true,"totalColumns":3,"maxResultSize":-1,"families":{"0":["\\x00\\x00\\x00\\x00","\\x80\\x0B","\\x80\\x0C"]},"caching":2147483647,"maxVersions":1,"timeRange":[0,9223372036854775807]}]]]"
iteratorFactory = {ParallelIteratorFactory$1@4328}
initFirstScanOnly = false
scans = {ArrayList@4329} size = 1
splits = {SingletonImmutableList@4330} size = 1
physicalTableName = {byte[4]@4331}
plan = {ScanPlan@3968}
scanId = "8f373aba-e360-94d0-ae67-3774e06c338e"
mutationState = {MutationState@4333}
scanGrouper = {DefaultParallelScanGrouper@3967}
allFutures = {ArrayList@4334} size = 0
estimatedRows = {Long@4335} 1
estimatedSize = {Long@4336} 206
estimateInfoTimestamp = {Long@4337} 0
hasGuidePosts = false
scan = {Scan@4249} "{"loadColumnFamiliesOnDemand":true,"startRow":"1","stopRow":"1\\x00","batch":-1,"cacheBlocks":true,"totalColumns":3,"maxResultSize":-1,"families":{"0":["\\x00\\x00\\x00\\x00","\\x80\\x0B","\\x80\\x0C"]},"caching":2147483647,"maxVersions":1,"timeRange":[0,9223372036854775807]}"
useStatsForParallelization = true
caches = {HashMap@3974} size = 0
dataPlan = null
context = {StatementContext@4269}
tableRef = {TableRef@4338}
groupBy = {GroupByCompiler$GroupBy$1@4339}
orderBy = {OrderByCompiler$OrderBy@4340}
hint = {HintNode@4341} "/*+ NO_INDEX */ "
limit = null
offset = null
openIterators = {ArrayList@4326} size = 0
index = 0
closed = false
plan = {ScanPlan@3968}
numParallelFetches = 0

这是最终返回的 iterator 对象。可以看出 Scan 对象的 startRow/stopRow 已经被替换成了 email=‘test@test.com’ 的 ID 值,也就是 1。到此为止也就通过索引表改写了源 SQL 的执行计划。

debug 分析到这里就结束了,仍然还有很多细节没有探讨清楚,感兴趣的读者可以自行 debug。

semi join 生效条件
https://blog.csdn.net/lppl010_/article/details/80301699

第七章

经过前面的分析,我们知道 PhoenixSQL 会经过优化器改写、优化,索引会被翻译成对应的索引表。接下来会介绍 SQL 各种不同的形式及其优化后的共同特点,以便扩展全文索引。

考虑到 SQL 编译的复杂性,以及自身精力的有限性,Phoenix 自身的执行优化引擎的源码不再分析,我会设计一些 SQL,用以探测优化后的执行 SQL 形式,由此来观察 Phoenix 查询索引的方式。

通过前面的文章我们知道,在 PhoenixStatement.executeQuery 方法通过 QueryOptimizer 对初步生成的执行计划进行了优化。

复制代码
plan = connection.getQueryServices().getOptimizer().optimize(PhoenixStatement.this, plan);

此处我们需要查看优化后的 plan 对应的 SQL,该如何查看呢?总不能下一个断点,每次都停到这里看一下吧。这里介绍一个小技巧,就是在不中断程序的情况下,打印调试代码的变量值。

在断点标志上面点击鼠标右键,会弹出会话框。可以看出居然还可以根据某个条件进行中断,不过这不是我们关注的点。

以调试形式,深入理解凤凰全局索引

点击 “More” 出现下面的对话框:

以调试形式,深入理解凤凰全局索引

上面有 “Evaluate and log” 可以填写我们要打印的内容,此处可以用上下文的变量,然后单击 done,然后 debug 执行我们的代码,可以看到执行的每个 SQL 都有对应的、优化后的 SQL。

以调试形式,深入理解凤凰全局索引

认真观察上面的输出,我们发现一个规律:所有能走索引的 SQL 都会用对应索引表的表名替代原表名,并且过滤条件也会透传给索引表。

可是这个规律说明什么呢?

这就意味着,查询外部索引时,所有的信息都已经有了!包括查询的字段,过滤条件,聚合条件等。我们只需要根据这些条件查询外部索引就行了。

总结

经过前面的分析,我们已经对 Phoenix 索引的创建、维护、使用有了简单的了解。接下来用思维导图为大家简单总结下:

以调试形式,深入理解凤凰全局索引

上面的图画的比较简单,但可以用来帮助读者分析实现全文检索的基本过程。其实还是有很多细节没有讲解清楚的。比如字段如何投射、聚合如何实现、排序如何实现,索引表的 ROWKEY 如何编码、事务表有没有特殊的地方、如何做单元测试、索引字段如何与 ES(SOLR) 进行映射、Phoenix 查询条件如何映射成 ES 查询代码、全文检索实现过程如何更加通用以便适配 ES 和 SOLR、如何度量全文检索的性能、创建索引时如何将属性透传给 ES。

具体实现的过程还是比较复杂的,另外 Phoenix 的代码质量并不是特别高,很多地方实现方式不统一,既要兼顾对源码的侵入性小,又要兼顾实现的通用型,作者还是着实下了一番功夫的。

作者介绍

吴少杰,爱好大数据生态的技术和框架,对数仓架构和实时计算比较熟悉,目前主要从事大数据开发和架构的工作

本文来自 DataFun 社区

原文链接

https://mp.weixin.qq.com/s?__biz=MzU1NTMyOTI4Mw==&mid=2247493714&idx=2&sn=6e598f59824d54497e043a89d5003059&chksm=fbd75a3ecca0d3288dcc23325a9e340e6945ed1ddc51f0fe2d2dcbc7d2867f18d20f0d1fac37&scene=27#wechat_redirect

评论

发布