源码解读：PolarDB-X中的窗口函数-阿里云开发者社区

作者：越寒

为什么需要窗口函数？

Window是一个常用且重要的功能，PolarDB-X作为一款分布式数据库，自然也支持了窗口函数。对于业务开发来讲，其可以大大简化业务SQL的设计，比如分组排序功能，如果支持窗口函数，则只需使用排序函数即可，例子如下。例：我现在有一张表，包含学生姓名，学生班级，学生成绩，现在请你帮我写一条SQL，实现对每个班级内的同学进行排名的需求？有窗口函数时：

SELECT 
 student_name, 
 class_name, 
 score, 
 DENSE_RANK() OVER (PARTITION BY class_name ORDER BY score DESC) AS rank
FROM student_scores
ORDER BY class_name, rank ASC;

无窗口函数时：就需要写比较复杂的SQL，感兴趣的同学可以自行尝试或者网上搜索一下如何写这样的SQL（或许这也可能在面试中被问到:)）。

窗口函数是什么？

本质上window是一种aggregation，但是不同于agg的是，agg要进行聚合的是该分组内的所有记录，每个分组也只会输出一行记录，而window则可以控制对于每一行来讲，想要聚合的记录到底是哪些，当然这种控制也是通过规则进行约束的，输出的记录行数等于输入的记录行数，下面贴了一张图，应该还比较清楚:)。

上图中的partitioin by与大家常写的SQL中的group by基本等价，比较容易理解，不再赘述。我们来展开介绍一下frame是个什么样的东西，如前所述，frame控制的是进行partition分区后，在该partition分区内，该行应该选择哪些行进行聚合运算。具体来讲，我们是通过between和and来指定我们希望框定向前和向后的哪些行进行聚合运算的，而指定方式也无非是行数，当前行以及不做限制，据此我们可以将frame分为四类，如下图所示。

实际上，将frame划分为不同的类型，可以指导我们根据不同的frame类型进行不同的优化，其主要目的是为了避免重复计算。比如对于unbounded preceding and unbounded following类型，显然我们只需对该分组进行一次计算即可，该分组的后续记录可直接使用该结果。而对于sliding frame则不能这样处理，其处理要复杂一些，对于每一行，我们都需要找出该行所对应的框定行，然后对这些行进行聚合运算。进一步的，frame可以分为两类，row模式和range模式，row模式寻找边界的依据是行数，而range模式的依据则是值。我们拿一个例子出来，看下row模式和range模式的区别吧。如下图所示，其frame定义均为between current row and current row，但是在row模式和range模式下，其选中的行并不相同。

在上述的介绍中，我们没有介绍每个分区内的order by字段，其实一个完整的window的定义包含partition by, order by和frame specification。但order by理解起来也比较简单，顾名思义，order by即指定对于每个分区内的行，应当按照什么顺序进行排序，frame中的向前向后多少行也是基于该排序后的集合。 Q：抛一个问题，有兴趣的朋友可以思考一下，向前向后的行一定是连续的，这是为什么呢？比如range模式下如何保证这一点？

窗口函数的设计与实现

窗口函数可能不是那么容易理解，所以我们在前面进行了比较多的介绍，现在我们终于来到了设计与实现部分。

如何执行窗口函数？

我们以一条SQL为例吧，如下所示，partition字段为c1，排序字段为c2，frame定义为rows between 1 preceding and 1 following。

select 
        c1, 
        c2, 
        avg(c2) over (
      partition by c1 
      order by c2 
      rows between 1 preceding and 1 following
    )
from t;
首先，关键点1，c1字段相同的记录应当被放置在一起（shuffle）；其次，关键点2，当我们对c1 + c2进行排序时，即可识别每行属于哪个分区以及该行的相关行是哪些。据此我们来展开介绍一下优化器和执行器的设计。

优化器

我们可以把优化器中的相关规则分为生成规则与优化规则，所谓生成规则，即用来确保window能够被正确的识别和转换，而优化规则则是为了优化某些场景下带有窗口函数的SQL。

生成规则

生成规则主要包含三条，project如何生成logical window (ProjectToLogicalProjectAndWindowRule )，logical window如何转换为执行时所需的sort window (LogicalWindowToSortWindowRule )，如何让sort window并行起来 (MppSortWindowConvertRule )。在project如何生成logical window 中，并没有太多要聊的东西，如何识别和转换可以直接查看相关源码，我们主要来聊一个有意思的问题，还是拿个SQL来举例子吧，如下。

select 
        c1, 
        c2, 
        avg(c2) over (
      partition by c1 
      order by c2 
      rows between 1 preceding and 1 following
    ),
    sum(c3) over (
      partition by c1 
      order by c2 
      rows between 1 preceding and 1 following
    )
from t;

Q：上述SQL应该生成几个window？ A：生成两个是最简单的，但其window的定义是相同的，所以理想情况下应该只需要生成一个window，window里面包含avg和sum函数就好了。代码如下所示，如果这两个window定义相同时，会被压到一个window中。

final List<Pair<RexWindow, Set<Integer>>> windowToIndices = new ArrayList<>();
for (int i = 0; i < exprs.size(); ++i) {
    final RexNode expr = exprs.get(i);
        if (expr instanceof RexOver) {
        final RexOver over = (RexOver) expr;
        // If we can found an existing cohort which satisfies the two conditions,
        // we will add this RexOver into that cohort
        boolean isFound = false;
        for (Pair<RexWindow, Set<Integer>> pair : windowToIndices) {
            if (pair.left.equals(over.getWindow())) {
                pair.right.add(i);
                isFound = true;
                break;
            }
        }
        // This RexOver cannot be added into any existing cohort
        if (!isFound) {
            final Set<Integer> newSet = Sets.newHashSet(i);
            windowToIndices.add(Pair.of(over.getWindow(), newSet));
        }
    }
}

Q：其实这里面有可供进一步优化的场景，本质上优化器和执行器需要更紧密的配合，感兴趣的朋友可以debug一下。接下来我们来看一下logical window如何转换为执行时所需的sort window。核心在于我们需要将sort的属性加入到window中，因为logical window本身是没有排序属性的，这里我们需要window的输入是按照partition column + sort column有序的，同时sort window也拥有此顺序。

RelCollation relCollation = RelCollations.EMPTY;
if (groupSets.cardinality() + orderKeys.size() > 0) {
    relCollation = CBOUtil.createRelCollation(sortFields, orderKeys);
}
// change trait set of input
RelNode newInput = convert(input, input.getTraitSet().replace(DrdsConvention.INSTANCE).replace(relCollation));
// change trait set of window
SortWindow newWindow =
SortWindow.create(
    window.getTraitSet().replace(DrdsConvention.INSTANCE).replace(relCollation),
    newInput,
    window.getConstants(),
    window.groups,
    window.getRowType(),
    window.getFixedCost()
);

细心的朋友会发现，我们在修改排序属性之外，还将window的convention修改为了DrdsConvention ，感兴趣的朋友可以思考一下为什么？此外，我们加入了排序属性之后，如果其input本身并不满足排序属性的要求时，是在哪里插入排序的算子的呢，答案是DrdsConvetion.enforce方法，相关代码如下。

RelCollation toCollation = required.getTrait(RelCollationTraitDef.INSTANCE);
RelDistribution toDistribution = required.getTrait(RelDistributionTraitDef.INSTANCE);
if (!RuleUtils.satisfyCollation(toCollation, input)) {
    RelTraitSet emptyTraitSet = input.getCluster().getPlanner().emptyTraitSet();
    MemSort memSort = MemSort.create(
    emptyTraitSet.replace(DrdsConvention.INSTANCE).replace(toCollation).replace(toDistribution),
    input,
    toCollation);
    return memSort;
} else {
    return input;
}

最后，我们来看一下如何将sort window并行起来，核心是我们现在要加上distribution的属性了，以便能够充分的并行，代码如下。

boolean noPartition = keys.size() == 0;
// for exchange(shuffle)
RelDistribution relDistribution = 
    noPartition ? RelDistributions.SINGLETON : RelDistributions.hash(groupSet);
RelCollation relCollation = 
    sortWindow.getTraitSet().getTrait(RelCollationTraitDef.INSTANCE);
input = convert(input, input.getTraitSet()
            .replace(MppConvention.INSTANCE)
            .replace(relDistribution)
            .replace(relCollation));
SortWindow newSortWindow = 
        sortWindow.copy(
        sortWindow.getTraitSet()
                    .replace(MppConvention.INSTANCE)
                    .replace(relDistribution),
        Arrays.asList(input)
        );

优化规则

所有的优化规则起码要回答三个问题，为什么能够优化，在哪些场景中能够使用，在能够使用的场景中这种优化是否总是正向的。相关的规则主要包括，ProjectWindowTransposeRule ，FilterWindowTransposeRule 和CBOJoinWindowTransposeRule 。 ProjectWindowTransposeRule用于将project尽可能下压到window下面，以便尽早过滤不需要的列，需要注意的是project中下层窗口函数计算结果的列显然不能推下去。

FilterWindowTransposeRule用于将filter尽可能下压到window下面，以便尽早过滤不需要的记录。这里面的核心有两个，首先，我们需要将filter中的condition分解为使用and连接的condition列表，比如c1 = 1 and (c2 > 1 or c2 > 2) -> List{c1=1, c2 > 1 or c3 > 2}。其次，对上述list中的condition进行循环判断，当window中用于partition的列需要包含该condition中的所有列时，该condition可以被推到window下面，否则不行，代码如下。

// decompose condition by AND
final List<RexNode> conditions =
    RelOptUtil.conjunctions(filterRel.getCondition());
for (RexNode condition : conditions) {
      ImmutableBitSet rCols = RelOptUtil.InputFinder.bits(condition);
            if (window.keys.contains(rCols)) {
        pushedConditions.add(condition.accept(new RelOptUtil.RexInputConverter(rexBuilder,
                origFields,
                window.getInput(0).getRowType().getFieldList(),
                adjustments)));
      } else {
        remainingConditions.add(condition);
      }
}

CBOJoinWindowTransposeRule 用于判断是否需要将join和window进行互换，准备来讲，其匹配的模式是join的右侧为filter，同时filter的输入是window，如下所示。

public static final CBOJoinWindowTransposeRule INSTANCE =
    new CBOJoinWindowTransposeRule(
      operand(LogicalJoin.class, 
              operand(RelNode.class, any()),
                    operand(LogicalFilter.class, operand(LogicalWindow.class, any()))
             ), 
      RelFactories.LOGICAL_BUILDER,
      "INSTANCE");

转换前后的子树结构如下图所示，并非对于所有SQL，右边的子树都更优，这里面的核心在于join的过滤性与filter的过滤性。如果join的过滤性非常好，则右边可能会更优，因为经过join后，输入到window中的记录数被大大削减。不过应用该规则需要比较小心，join的左边列必须要能保证全局唯一，否则经过join后，输入到window中的相关的右表的记录数会被放大，这就不满足原始的语义了，同时join应该为等值join并且join key与window的partition key相同。

执行器

window算子接收到的输入是按照partition key + sort column排好序的，所以我们要做的就是，找出每行记录对应的分区，然后根据目前缓存的记录和frame的定义，计算能够计算的所有行，如果需要输出，则向上层吐数据，否则继续接受下一批输入。当然，由于情况还比较多，所以有一些细节需要考虑，具体可参考OverWindowFramesExec ，也可对照下图进行理解。

其次是如何接入异步执行框架，因为这并不是一个通用的需求，并且我们还没有进行异步执行框架的源码解读，展开介绍不容易讲清楚，意义也不太大。最后，我们来聊两个细节的优化吧。第一个优化的出发点是针对特殊场景进行优化，即是否在所有场景下，我们都需要缓存数据，实际上只要窗口不会包含当前行的后续行，就不需要缓存数据，详情可见下图，这部分的处理在NonFrameOverWindowExec 算子中。

第二个优化的出发点是尽量避免当窗口滑动时，窗口函数需要全部重新计算。如下图所示，当我们从左边的窗口滑动至右边的窗口时，我们只需要把新增的记录放入窗口函数中计算即可，不必全部重新计算。

我们再来看一个稍微更复杂一点的滑动窗口，此时并非只有新增的记录，也有移除的记录，做增量的计算就变得更加复杂了。如下图所示，窗口中新增了y，但是移除了x，在这种情况下是否需要全量的重新计算取决于聚合函数的类型，比如sum是可以的，但是bit_and或者max之类的就是不行的，或者说没有那么容易，而且在这种情况下这种优化的效果是否有比较好的效果取决于窗口的大小。更加复杂一点的优化，感兴趣的朋友可以搜索线段树。