window function的实现

手机用户2895 2023-10-27

352

Window function 的语义

在一个 SELECT 语句里，window function 作用于 GROUP BY、HAVING之后，SELECT DISTINCT、ORDER BY、LIMIT 之前，其主要作用是，对输入的每一行数据，设定一个包含前后数据的“窗口”，在这个窗口内计算聚合函数以及一些特定的窗口函数的值。

这里通过一个例子展示窗口函数的用途。

mysql> SELECT
    ->    year, country, product, profit,
    ->    SUM(profit) OVER(PARTITION BY country) AS country_profit
    -> FROM sales
    -> ORDER BY country, year, product, profit;
+------+---------+---------+--------+----------------+
| year | country | product | profit | country_profit |
+------+---------+---------+--------+----------------+
| 1999 | FINLAND | P0      |    100 |            300 |
| 2000 | FINLAND | P1      |    100 |            300 |
| 2001 | FINLAND | P2      |    100 |            300 |
| 1999 | UK      | P0      |    100 |            100 |
| 1999 | USA     | P0      |    100 |            200 |
| 2000 | USA     | P1      |    100 |            200 |
+------+---------+---------+--------+----------------+

上面的 SELECT 获取了 sales 表里面的所有数据，并通过窗口函数 SUM(profit) OVER(PARTITION BY country) 为每一行添加了“当前这一行所对应的 country 的 total profit”，也就是说，这个窗口“覆盖”了当前这一行所对应 country 的所有数据；

为了避免 window function / window operator / window expression / window 等称呼的歧义，我们把

SUM(profit) OVER (PARTITION BY country) 这样的 clause 称为 window function（窗口函数），它的实现称为 window operator

SUM(profit) 这样的表达式称为 window expression （window 表达式）

(PARTITION BY country) 这样一个窗口的定义称为 window（窗口）

在窗口函数里，如何定义“窗口”，是关键。一个窗口的定义包含“三要素”（partition、order、frame）：

window_spec:
    [window_name] [partition_clause] [order_clause] [frame_clause]

比如：

(PARTITION BY column_set   ORDER BY column_set   ROWS BETWEEN ... )
或者
(PARTITION BY column_set   ORDER BY column_set   RANGE BETWEEN ...)

用户通过

PARTITION BY 定义如何对输入数据进行分区；如果没有定义则所有输入数据成为一个分区；
ORDER BY 定义每个分区里的数据如何排序；如果没有定义则默认不排序；
ROWS BETWEEN / RANGE BETWEEN 定义窗口起始和结束位置，也称为一个 “frame”；如果没有定义 frame 且没有 order by 则默认的 frame 是整个分区；如果没有定义 frame 但是定义了 order by 则默认的 frame 是分区第一行到 current row；需要注意，每一行的 frame 都不能越过这一行所在的分区；

下图中展示了窗口的几个概念：首先将输入数据按照 partition by 进行分区，然后按照 order by 对每个分区的数据进行排序，最后通过 ROWS BETWEEN 或者 RANGE BETWEEN 为当前行指定窗口的起始和结束；图中灰色的一行就是当前行（current row），当窗口滑动到下一行时，下一行变成当前行，窗口的起始和结束也随之变化；

window-spec.drawio.svg

其中，frame 的定义有两种方式

ROWS BETWEEN 选择前后几行；比如 ROWS BETWEEN 3 PRECEDING AND 3 FOLLOWING 表示往前 3 行到往后 3 行，一共 7 行数据（如果前面或者后面遇到了 partition 的边界，则少于 7 行）
RANGE BETWEEN 根据数据范围选择前后几行；比如 RANGE BETWEEN 3 PRECEDING AND 3 FOLLOWING 表示所有值在 [val_−3, val_+3] 这个范围内的行，val 为当前行的值（同样地，也不能越过 partition 的边界）

Untitled Diagram2.drawio.svg
通过上述窗口的定义，用户就可以通过

SELECT
	col0,
  col1,
  window_expr0(...) over (window0),
  window_expr1(...) over (window1)
FROM
	tables;

为 tables 中的每一行增加一个作用于 window0 的 window_expr0 的值，以及作用于 window1 的 window_expr1 的值；MYSQL 支持的所有窗口表达式在这里可以找到；另外，大多数的聚合函数都可以用作窗口表达式。

不难看出，窗口函数与 aggr/groupby 操作的区别是：aggr / groupby 操作会将多行聚合为一行或几行，而窗口函数不改变数据的行数，只会为每一行增加一个或多个窗口表达式的值。

窗口函数只能出现于 SELECT 以及 ORDER BY clause 中。

关于窗口函数，可以参考 MySQL 官方文档对 window function 的描述：https://dev.mysql.com/doc/refman/8.0/en/window-functions.html

这篇 paper（的前几个章节）也是不错的参考：Efficient Processing of Window Functions in Analytical SQL Queries

MySQL 对 window function 的支持

SQL 标准里面定义的 window function 的语法比较灵活，然而 mysql （以及其他一些商业数据库）并不完全支持所有的 window function 语义[MYSQL LIMITATION AND RESTRICTIONS]；这意味着 IMCI 不需要实现所有的 window function 的语义，能够简化我们的设计；

在 MySQL 不支持的相关特性中，下列三条是比较重要的：

aggregate window expression 中，不能带有 DISTINCT；i.e., 不支持 SUM(DISTINCT col0) OVER (window) 这样的语义；
不支持根据 current row 的值动态指定 frame 的边界；i.e., 不支持 ORER BY col ROWS BETWEEN **col **- 1 AND col + 1; 这样的窗口；
不支持嵌套的 window function

简单地对比一下其他数据库对 window function 的支持：

SQL SERVER
SQL SERVER 也不支持上面 3 条特性；总体上来说，SQL SERVER 对 window function 的支持度甚至没有 mysql 高 [SQL SERVER LIMITATION AND RESTRICTIONS]。

SQL SERVER 甚至只能在 RANGE BETWEEN 中使用 UNBOUNDED 和 CURRENT ROW 来指定 frame 的起始和结束i.e., 不支持 RANGE BETWEEN 1 PRECEDING AND CURRENT ROW 这样的操作（这基本上就等同于不支持 RANGE BETWEEN 这种模式了，因为可以用 ROWS BETWEEN 来表达同样的语义）；这个约束使得它的所有 frame 的起始和结束位置都与 current row 的数据没有关系，大大简化了它对 window function 的实现；

实际上，SQL SERVER 里面甚至没有一个 window operator，它只增加了一个“window spool”即可实现 window function；

POSTGRESQL
从文档和使用体验来看，PG 对 window function 的支持程度跟 MySQL 差不多；同样不支持上面 3 条特性；

ORACLE
从文档和使用体验来看，相比于上述三款数据库，oracle 对 window function 的支持程度是最高的；支持 aggregate window function 中带有 DISTINCT；支持根据 current row 的值动态指定 frame 的边界；不过也不支持嵌套的 window function；

SQL SERVER 中 window function 的实现

由于上述约束，SQL SERVER 对 window function 的支持程度没有 mysql 高；但是因为有了“窗口的起始和结束位置都与 current row 的数据没有关系”这个约束， SQL SERVER 对 window function 的实现非常简洁，对 IMCI的实现有一定的参考意义，因此这里做一个简单描述。

不带 frame 的 window function

在没有指定 frame 的时候（e.g., SUM(profit) OVER(PARTITION BY country))，window expression 的作用范围是整个 partition，因此 window function 的实现相对简单，只要将输入数据分区，然后计算 window expression 即可；例如下面的 SQL：

SELECT 
  year, 
  country, 
  product, 
  profit, 
  SUM(profit) OVER(PARTITION BY country) AS country_profit 
FROM 
  sales;

在 SQL SERVER 里面，执行计划如下：

并行的执行计划加了 exchange:

其中比较关键的两个算子是 Segment 和 Table Spool：

Segment 用于对输入数据进行分区，分成不同的“segment”；在上面的 query 里面，会按照 window 里面的 country 这一列进行分区；这个 Segment 类似于 IMCI 的 HashJoin 和 HashGroupby 里面的 partition，区别在于这个 Segment 算子依赖于前面的 Sort 算子做“streaming”的分区，而 IMCI 的 partition 直接用 hash 的方式进行分区，速度上会有一些优势；
Table Spool 用于“cache”输入的数据，类似于 IMCI 的 Materialize 算子，区别在于：
1. table spool 有 lazy spool 和 eager spool 的区别，lazy spool 只会被动 cache 那些“流经它的数据”；
2. spool 下面如果是 segement 算子，则会有“segment spool”的行为，下面会描述；
3. spool 算子有 primary spool 和非 primary spool 的区分；primary spool 是真正保存数据的，非 primary spool 则是对 primary spool 的引用；上图中，上面的 spool 是 primary spool，下面两个 spool 是非 primary spool；区分这两种 spool 的一个用处是，可以保持一个树状的执行计划，而不是有 common child 的 DAG；比如一个 query 如果在多个地方引用了同一个 VIEW，那么只需要一个地方物化这个 VIEW（i.e., primary spool），其他地方引用这个 spool 就行（非 primary spool）；又比如在子查询的处理中，也可以这么处理 shared sub-plan；

这个执行计划的执行过程是：

table scan -> sort -> segment 首先将数据分区
table spool 的每次 fetch from segment，都会取一个完整的 segment
顶层的 nested loop join 驱动下面 nested loop join 执行

下面的 nested loop join 的 predicate 是 true，其子节点分别
1. 从 table spool 中取每个 segment （i.e., 每个 partition）的数据，进行 stream aggregate（i.e., SUM(profit)）；
2. 从 table spool 中取每个 segment
  
  将两个子节点 join 起来，就可以得到一个 partition 的结果集；

需要说明的是：

真正对每个 partition 进行运算的是第二个 nested loop join，第一个 nested loop join （i.e., SELECT 后面那个）只是为了驱动 table scan -> sort -> segment -> table spool 这条 pipeline，将每个 segment 保存到 spool 里面；
这里的 table spool 下面挂着一个 segment 算子，因此每次 cache 都只是 cache 一个“分区”

带 frame 的 window function

在指定 frame 的 window function 里面，当窗口移动到下一行，窗口的范围都会随之改变，因此如何在这种情况下快速计算不同窗口的值，就需要一定技巧。

与 MySQL 不同的是，在 SQL SERVER 的 window function 里面，“窗口的起始和结束位置都与 current row 的数据没有关系”，也就是说，

窗口是定长的，或者
窗口从 partition 的第一行/最后一行开始不断扩张/缩小（e.g, ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)

因此在带 frame 的 window function 里面，SQL SERVER 使用一个 “window spool”来实现。例如下面的 SQL：

SELECT 
  year, 
  country, 
  product, 
  profit, 
  SUM(profit) OVER(
    PARTITION BY country 
    ORDER BY profit
    ROWS BETWEEN 3 PRECEDING and CURRENT ROW   --- 窗口范围是前三行到当前行
  ) AS country_profit0 
FROM 
  sales;

在这个例子里面，窗口的访问是前三行到当前行，其执行计划是：

其中的 window spool 的主要作用是：

将 segment 中的每个分区 cache 住（与上文的 table spool 类似）；
在往上吐数据时，为每个 window 的数据增加一个 “WindowCount”，使得同一个 window 中的行拥有相同的 WindowCount，可以被聚合到一起；

例如，对于如下属于同一个分区的数据：

+------+---------+---------+--------+
| year | country | product | profit |
+------+---------+---------+--------+
| 1999 | FINLAND | P0      |    100 |
| 2000 | FINLAND | P1      |    100 |
| 2001 | FINLAND | P2      |    100 |
| 2002 | FINLAND | P3      |    100 |
| 2003 | FINLAND | P4      |    100 |
| 2004 | FINLAND | P5      |    100 |
+------+---------+---------+--------+

Window spool 向上吐数据时，其中一个方式是：

第 1 次吐数据：
+------+---------+---------+--------+-------------+
| year | country | product | profit | WindowCount |
+------+---------+---------+--------+-------------+
| 1999 | FINLAND | P0      |    100 | 1           |
+------+---------+---------+--------+-------------+

第 2 次吐数据：
+------+---------+---------+--------+-------------+
| year | country | product | profit | WindowCount |
+------+---------+---------+--------+-------------+
| 1999 | FINLAND | P0      |    100 | 2           |
| 2000 | FINLAND | P1      |    100 | 2           |
+------+---------+---------+--------+-------------+

第 3 次吐数据：
+------+---------+---------+--------+-------------+
| year | country | product | profit | WindowCount |
+------+---------+---------+--------+-------------+
| 1999 | FINLAND | P0      |    100 | 3           |
| 2000 | FINLAND | P1      |    100 | 3           |
| 2001 | FINLAND | P2      |    100 | 3           |
+------+---------+---------+--------+-------------+

第 4 次吐数据：
+------+---------+---------+--------+-------------+
| year | country | product | profit | WindowCount |
+------+---------+---------+--------+-------------+
| 1999 | FINLAND | P0      |    100 | 4           |
| 2000 | FINLAND | P1      |    100 | 4           |
| 2001 | FINLAND | P2      |    100 | 4           |
| 2002 | FINLAND | P3      |    100 | 4           |
+------+---------+---------+--------+-------------+

第 5 次吐数据：
+------+---------+---------+--------+-------------+
| year | country | product | profit | WindowCount |
+------+---------+---------+--------+-------------+
| 2000 | FINLAND | P1      |    100 | 5           |
| 2001 | FINLAND | P2      |    100 | 5           |
| 2002 | FINLAND | P3      |    100 | 5           |
| 2003 | FINLAND | P4      |    100 | 5           |
+------+---------+---------+--------+-------------+

（当然，也可以把这个 5 次合并成一次）
由于每个 window 的 WindowCount 是相同的，因此上面的 stream aggr 只需要 group by WindowCount 就可以得到正确的结果。

不难看出，因为做了重复运算，因此这里的复杂度是 O(N * M)，N 是数据量，M 是窗口长度。

对多个 window function 的处理

当一个 query block 里面出现多个 window function 时，e.g.,：

SELECT
	col0,
  col1,
  window_expr0(...) over (window0),
  window_expr1(...) over (window1)
FROM
	tables;

SQL SERVER 的处理方法是：

当多个 window function 里面的 window 定义相同时，做合并（e.g., 使用同一个 window spool）；
当多个window function 里面的 window 定义不同时（e.g., partition columns 不同，或者 frame 的定义不同），在一个 window spool 上叠加另一个 window spool。

其他数据库中 window function 的实现

window function 三要素（partition、order、frame）中 partition 、sort 的处理都比较通用（要么是排序要么是哈希，同时考虑上并行处理时的如何对数据分区），因此这里忽略。

不同数据库对 frame 的处理却不太一样，这里简单描述一下：

POSTGRESQL
在窗口移动的过程中保存一个 “running aggregation”，因此对于那种不断扩张的窗口定义（e.g., range between unbounded preceding and current row），有加速作用；但是对于定长的移动窗口（e.g., sum(b) over (order by a rows between 5 preceding and 5 following)），作用不大；
(注：从 paper 里面读的，没有读代码）

MYSQL
从 mysql80 的 sql/sql_executor.cc 文件的里 process_buffered_windowing_record() 这个函数上方的长注释大概可以看出 mysql 的处理和优化逻辑：

对于 range between unbounded preceding and current row 这种可以用 running aggregation 的情况，使用 running aggregation；
对于 sum(b) over (order by a rows between 5 preceding and 5 following) 这种定长窗口移动，且 window expression 是 SUM / AVG 等函数时，可以在窗口移动的时候，把上一个窗口的边界值去掉（i.e, Removable Cumulative Aggregation）；这种优化在某些情况下会有精度问题，因此可以通过 windowing_use_high_precision 这个开关选择是否打开；
对于其他的情况，利用一个 window buffer，在窗口移动的时候将数据拷到这个 buffer 里面，然后算 window expression；这种算法 worst case 的复杂度还是 O(n * m), n 是数据量，m 是窗口的最大长度；

看起来 mysql 的优化还是比较到位的。

ORACLE
暂时没找到相关资料。

HYPER
对于常见窗口定义，维持“running aggregation”，或者 “Removable Cumulative Aggregation”，来快速得到window expression 的值；对于其他情况，利用一个线段树（segment tree）来保证无论窗口如何变化，使得取某一个窗口的 window expression 值，最多只需 log n 的时间，因此 worst case 的性能是 O(n log(n))；参考《Efficient Processing of Window Functions in Analytical SQL Queries》这篇 paper。
这种方式既不会影响一般情况下的性能，又能够限制 worst case 的复杂度，是比较好的处理方式。

IMCI 的 window function 设计&实现

模仿 SQL SERVER 的处理方式；
为了跟 mysql 完全兼容，增加一个 window operator 处理那些“窗口的范围随着 current 的数据而变化”的case;考虑利用上 segment tree，控制 worst case 的复杂度；

我们首先实现 1.，快速满足大部分用户的需求；后续再增加 2.，做到完全兼容 mysql。

要实现 1. ，我们只需要增加 segment 、table spool 以及 window spool 算子：

IMCI 的 HashJoin 和 HashGroupby 里面的 partition 过程单独抽出来，就是一个 segment 算子；
实现一个 table spool 算子，这个算子后续也可以用于处理子查询；
实现一个 window spool 算子

这都是相对简单的 task。

polardb

「喜欢这篇文章，您的关注和赞赏是给作者最好的鼓励」

关注作者