分页的limit_Presto分页功能概述

引言

本文假设读者已经熟悉了 Presto QE 执行模型的一些基本概念，比如 Statement、Query、Stage、Task、Split、Driver、Operator、Exchange。

当前社区版的 Presto并不能很好的支持分页语法，究其原因，我的理解是因为当 offset 过大会造成性能的损失，假设 offset 1,000,000 limit 20，则数据库会扫描 1,000,020 行后然后把前一百万行数据给丢掉，该操作的成本比较高，如图1为一个测试结果【具体见文章末尾参考】，当随着offset增大的时候，Query 的执行时间也会跟着变得越来越大。

图1

该篇文章以如何实现分页语法功能做为一个开篇，来介绍如何实现一个分页功能在当前 Presto中。在具体开始开始前，先看下 MySQL 和 PostgreSQL 的分页语法。

MySQL分页语法

MySQL 里面提供的分页语法如下，而用户在写SQL的过程中 (比如 limit 3,5 )，该语法由于没有精确的指定Limit 后面的数字到底代表什么含义而显得有些歧义。

Limit offset, limit

PostgreSQL分页语法

PostgreSQL 提供的Limit分页语法如下：

LIMIT { count | ALL } OFFSET start

Presto当前的语法

只能提供 Limit 后加一个参数的语法结构。关于 sql 的语法定义在 SqlBase.g4文件中，可以看到如下的 Limit 语法并不支持偏移量这样的功能。

query:  with? queryNoWith;
queryNoWith:queryTerm(ORDER BY sortItem (',' sortItem)*)?(LIMIT limit=(INTEGER_VALUE | ALL))?;

假设我们要给 Presto 的分页语法如下，也即是未来给用户写sql语句的样子，注意这里只是自己定义的一种语法格式，用户也可以随意指定想要的语法格式。

offset x limit y

要让Presto执行引擎能够识别该语法，则首先需要在 SqlBase.g4层面支持这样的语法结构，然后进行词法分析、语法分析最后生成AST树。ANTLR会根据 SqlBase.g4 生成其中 SqlBaseListener 和 SqlBaseVisitor 两个文件。SqlBaseListener 会用在一些错误的汇报中，而 SqlBaseVisitor 则用来生成 Presto 的抽象语法树(Node Hierarchy)，具体生成AST的工作在 com.facebook.presto.sql.parser.AstBuilder#visitXXX 完成，下面的这段代码则为访问Query时产生的Node。

class AstBuilder extends SqlBaseBaseVisitor<Node>@Override
public Node visitQuery(SqlBaseParser.QueryContext context)
{Query body = (Query) visit(context.queryNoWith());return new Query(getLocation(context),visitIfPresent(context.with(), With.class),body.getQueryBody(),body.getOrderBy(),body.getOffset(),body.getLimit());
}

新的语法文件如下，增加 OFFSET 关键字的支持。

queryNoWith:queryTerm(ORDER BY sortItem (',' sortItem)*)?(OFFSET offset=INTEGER_VALUE)?(LIMIT limit=(INTEGER_VALUE | ALL))?;

LimitNode & LimitOperator

LimitNode为逻辑执行计划节点，主要是用来执行Limit操作。LogicalPlanner 会针对生成的执行计划进行优化，比如LimitPushDown用于将Limit条件进行下推，从而减少后续节点处理的数据量，提高执行效率。

LimitOperator 进行处理具体的数据，属于物理执行层面。接受以Page做为输入源，然后输出处理后的Page到下游的Operator。支持分页功能的具体操作层面的改动就在LimitOperator算子里，需要增加处理 offset 参数的LimitOperator。此外还有一种实现思路就是不再原有的Operator层面进行改动，而当Presto发现含有offset关键字的时候自动插入一个新的Operator，比方我们叫做 OffsetOperator，在LimitOperator 之前进行过滤，从而实现松耦合。

当前 LimitOperator 的实现，不支持分页，addInput里面为算子的具体实现逻辑。

public void addInput(Page page){checkState(needsInput()); // 检查是否还需要数据// 如果当前的page包含的行数小于等于需要的数据 remainingLimit，则重置当前还需要的行数// 当前输出的page即为输入进来的page，这是显然的if (page.getPositionCount() <= remainingLimit) {remainingLimit -= page.getPositionCount();nextPage = page;}else {// blocks 即为有多少列数据，一个Block在Presto里面为一列Block[] blocks = new Block[page.getChannelCount()];for (int channel = 0; channel < page.getChannelCount(); channel++) {Block block = page.getBlock(channel);// 构造每一列数据blocks[channel] = block.getRegion(0, (int) remainingLimit);}// 构造当前的最后一个输出PagenextPage = new Page((int) remainingLimit, blocks);remainingLimit = 0;}}

支持分页实现的核心逻辑

    public void addInput(Page page){checkState(needsInput());// 如果当前的偏移量大于当前page的行数，则重置偏移量if (offset >= page.getPositionCount()) {offset -= page.getPositionCount();}else {if (offset == 0 && remainingLimit >= page.getPositionCount()) {nextPage = page;remainingLimit -= page.getPositionCount();}else {// 当前page还剩余多少行元素long remainingPosition = (int) (page.getPositionCount() - offset);// 如果用户需要的limit的数量大于等于当前page剩余的行数，则当前page只能提供remainingPosition行// cntInPage为当前page提供的行数, 通常情况下, remainingLimit >= remainingPosition = truelong cntInPage = remainingLimit >= remainingPosition ? remainingPosition : remainingLimit;Block[] blocks = new Block[page.getChannelCount()];for (int channel = 0; channel < page.getChannelCount(); channel++) {Block block = page.getBlock(channel);blocks[channel] = block.getRegion((int) offset, (int) cntInPage);}nextPage = new Page((int) cntInPage, blocks);// 用户还需要多少limit元素，即 offset X limit Y 中的 Y 的个数remainingLimit -= cntInPage;// 把当前的偏移量置成 0offset = 0;}}}

分布式执行分析

假设待执行的SQL为

select col1 from table offset 2 limit 2

用户的数据表如图2。在 Stage1 的时候，LimitOperator 的 Driver 数为4个，假设此时数据有4个Splits，在分布式阶段的 partial 阶段，数据被切分成4份，每个Split被一个Driver进行计算。

图2

图3

如果此时代码针对4个Splits的数据分别执行 offset 2 limit 2 则会产生如图3的问题，即每个split 不会向上游的算子产生数据，因为每个分片只有一条数据，而此时 offset 的值已经为2。怎么解决该问题？如图4为解决partial阶段问题的草图，把 offset 2 limit 2 在 partial 阶段 rewrite 成 offset 0 limit 4 则可。

图4

到了 final 阶段，LimitOperator 只会在一台机器上进行汇总，由于分页语法不要求语义上保证有序，所以最终向stage0输出数据的时候每次结果集并不是固定的，这点和TopNOperator实现所有不同，TopNOperator的实现在此不进行展开。如图5，在最后的 final 阶段，继续执行 offset 2 limit 2 语义，输出最终的结果返回给用户，理论上分析，执行同样sql输出的结果集每次都不同，也不保证有序。

图5

在 partial 和 final 阶段 LimitOperator 的构造过程是不同的，代码如下

 public Operator createOperator(DriverContext driverContext){if (partial) {// partial 阶段，进行 rewritereturn new LimitOperator(operatorContext, types, 0, offset + count);} else {// final 阶段return new LimitOperator(operatorContext, types, offset, count);}}

效果

以 memory 为 connector 进行测试。图6为用户的表，图7，8为执行的2次查询，可以看出每次查询出的结果集并不相同，顺序也不一致（逆序、顺序）和理论分析保持一致。

图6

图7

图8

结语

该文章以新增分页语法为介绍，以 LimitOperator 算子为具体实现。介绍了如何给 Presto 增加新的语法规则以及在物理执行层Operator的改造，最后以分布式执行分析和测试结果结束整篇文章，希望能够给大家一些帮助，欢迎大家讨论和留言，谢谢。

参考

https://www.eversql.com/faster-pagination-in-mysql-why-order-by-with-limit-and-offset-is-slow/www.eversql.com

彩蛋

稷和：火山执行模型 VS Presto QE 模型