如何为PostgreSQL创建自定义窗口函数？(运行平均值示例)

Question

我真的想更好地理解在PostgreSQL中创建一个在windows上运行的UDF所涉及的内容.我做了一些关于如何创建UDF的搜索,但没有找到如何在窗口上操作的示例.

为此,我希望有人愿意分享如何编写UDF(可以是C,pl/SQL或PostgreSQL支持的任何过程语言)的代码,以计算窗口中数字的运行平均值.我意识到有一些方法可以通过应用标准的平均聚合函数和窗口语法(我相信语法之间的行)来实现这一点,我只是要求这个功能,因为我认为这是一个很好的简单例子.此外,我认为如果存在平均函数的窗口版本,那么数据库可以保持运行总和和观察计数,并且在每次迭代时不会总结几乎相同的行集.

Answer 1

你必须查看postgresql源代码postgresql/src/backend/utils/adt/windowfuncs.c和postgresql/src/backend/executor/nodeWindowAgg.c

没有好的文档:( - 全功能窗口功能只能在C或PL/v8中实现 - 没有其他语言的API.

http://www.pgcon.org/2009/schedule/track/Version%208.4/128.en.html PostgreSQL实施作者的演讲.

我发现只有一个非核心实现 - http://api.pgxn.org/src/kmeans/kmeans-1.1.0/

http://pgxn.org/dist/plv8/1.3.0/doc/plv8.html

Answer 2

根据文档“用户可以添加其他窗口函数。此外，任何内置或用户定义的普通聚合函数都可以用作窗口函数。 ”（第 4.2.8 节）。这对我计算股票分割调整有用：

CREATE OR REPLACE FUNCTION prod(float8, float8) RETURNS float8
  AS 'SELECT $1 * $2;'
  LANGUAGE SQL IMMUTABLE STRICT;

CREATE AGGREGATE prods ( float8 ) (
  SFUNC = prod,
  STYPE = float8,
  INITCOND = 1.0
);

create or replace view demo.price_adjusted as
  select id, vd,
    prods(sdiv) OVER (PARTITION by id ORDER BY vd DESC ROWS UNBOUNDED PRECEDING) as adjf,
    rawprice * prods(sdiv) OVER (PARTITION by id ORDER BY vd DESC ROWS UNBOUNDED PRECEDING) as price
  from demo.prices_raw left outer join demo.adjustments using (id,vd);

这是两个表的模式：

CREATE TABLE demo.prices_raw (
  id VARCHAR(30),
  vd DATE,
  rawprice float8 );

CREATE TABLE demo.adjustments (
  id VARCHAR(30),
  vd DATE,
  sdiv float);

Answer 3

从表开始

付款
+------------------------------------------+
| 客户 ID | 金额 | 项目 |
| 5 | 10 | 10 书 |
| 5 | 71 | 71 鼠标|
| 7 | 13 | 封面|
| 7 | 22 | 22 电缆|
| 7 | 19 | 19 书 |
+------------------------------------------+

SELECT customer_id, 
    AVG(amount) OVER (PARTITION BY customer_id) AS avg_amount,   
    item, 
FROM payments`

我们得到

+----------------------------------+
| 客户 ID | 平均金额 | 项目 |
| 5 | 40.5 | 40.5 书 |
| 5 | 40.5 | 40.5 鼠标|
| 7 | 18 | 18 封面|
| 7 | 18 | 18 电缆|
| 7 | 18 | 18 书 |
+----------------------------------+

AVG作为聚合函数，它可以充当窗口函数。然而，并非所有窗口函数都是聚合函数。聚合函数是不复杂的窗口函数。

在上面的查询中，我们不使用内置AVG函数并使用我们自己的实现。做同样的事情，只是由用户实现。上面的查询变成：

SELECT customer_id, 
    my_avg(amount) OVER (PARTITION BY customer_id) AS avg_amount,   
    item, 
FROM payments`

与之前的查询的唯一区别是AVG已替换为my_avg. 我们现在需要实现我们的自定义函数。

关于如何计算平均值

将所有元素相加，然后除以元素数量。对于customer_id7 来说，那就是(13 + 22 + 19) / 3 = 18。我们可以将其划分为：

• 一步一步的积累——总和。
• 最后一个操作——除法。

关于聚合函数如何得出结果

平均值是分步计算的。只有最后一个值是必需的。从初始值 0 开始。

1. Feed 13. 计算中间/累加和，即 13。
2. Feed 22. 计算累计总和，需要之前的总和加上此元素：13 + 22 = 35
3. Feed 19. 计算累计总和，需要之前的总和加上这个元素：35 + 19 = 54。这是需要除以元素数量 (3) 的总数。
4. 步骤 3. 的结果被输入到另一个函数，该函数知道如何将累积和除以元素数量

这里发生的情况是，状态从初始值 0 开始，每一步都发生变化，然后传递到下一步。

只要有数据，状态就会在步骤之间移动。当所有数据被消耗时，状态进入最终函数（终端操作）。我们希望状态包含累加器以及终端操作所需的所有信息。

在计算平均值的特定情况下，终端操作需要知道累加器使用了多少个元素，因为它需要除以该值。因此，状态需要包括累积和和元素数量。

我们需要一个包含两者的元组。预定义的POINTPostgreSQL 类型可以解决这个问题。POINT(5, 89) 表示值为 89 的 5 个元素的累加和。初始状态将为 POINT(0,0)。

累加器是在所谓的状态函数中实现的。终端操作是在所谓的“最终函数”中实现的。

定义自定义聚合函数时，我们需要指定：

• 聚合函数名称和返回类型
• 初始状态
• 基础设施将在步骤之间传递到最终函数的状态类型
• 状态函数——知道如何执行累积步骤
• 最终函数——知道如何执行终端操作。并不总是需要（例如，在 SUM 的自定义实现中，累加和的最终值就是结果。）

这是自定义聚合函数的定义。

CREATE AGGREGATE my_avg (NUMERIC) ( -- NUMERIC is what the function returns
    initcond = '(0,0)', -- this is the initial state of type POINT
    stype = POINT, -- this is the type of the state that will be passed between steps
    sfunc = my_acc, -- this is the function that knows how to compute a new average from existing average and new element. Takes in the state (type POINT) and an element for the step (type NUMERIC)
    finalfunc my_final_func -- returns the result for the aggregate function. Takes in the state of type POINT (like all other steps) and returns the result as what the aggregate function returns - NUMERIC 
);

剩下的唯一事情就是定义两个函数my_acc和my_final_func。

CREATE FUNCTION my_acc (state POINT, elem_for_step NUMERIC) -- performs accumulated sum
RETURNS POINT
LANGUAGE SQL
AS $$
    -- state[0] is the number of elements, state[1] is the accumulated sum
    SELECT POINT(state[0]+1, state[1] + elem_for_step);
$$;

CREATE FUNCTION my_final_func (POINT) -- performs devision and returns final value
RETURNS NUMERIC
LANGUAGE SQL
AS $$
    -- $1[1] is the sum, $1[0] is the number of elements
    SELECT ($1[1]/$1[0])::NUMERIC;
$$;

现在上面定义的函数已经可用，CREATE AGGREGATE将成功运行。现在我们已经定义了聚合，可以运行基于my_avg而不是内置的查询：AVG

SELECT customer_id, 
    my_avg(amount) OVER (PARTITION BY customer_id) AS avg_amount,    
    item, 
FROM payments`

结果与使用内置AVG.

PostgreSQL 文档建议用户仅限于实现用户定义的聚合函数：

除了这些[预定义窗口]函数之外，任何内置或用户定义的通用或统计聚合（即非有序集或假设集聚合）都可以用作窗口函数；

我怀疑的ordered-set or hypothetical-set aggregates意思是：

• 返回的值与所有其他行相同（例如AVG和SUM。相反，RANK根据更复杂的条件为组中的所有行返回不同的值）
• 分区时 ORDER BY 没有任何意义，因为无论如何，所有行的值都是相同的。相反，我们想要ORDER BY在使用时RANK()

询问：

SELECT customer_id, item, rank() OVER (PARTITION BY customer_id ORDER BY amount desc) FROM payments;

几何平均数

以下是用户定义的聚合函数，我发现没有内置聚合，可能对某些人有用。

状态函数计算各项自然对数的平均值。

最终函数将常量提升e为累加器提供的任何值。

CREATE OR REPLACE FUNCTION sum_of_log(state POINT, curr_val NUMERIC)
RETURNS POINT
LANGUAGE SQL
AS $$
    SELECT POINT(state[0] + 1,
        (state[1] * state[0]+ LN(curr_val))/(state[0] + 1));
$$;

CREATE OR REPLACE FUNCTION e_to_avg_of_log(POINT)
RETURNS NUMERIC
LANGUAGE SQL
AS $$
    select exp($1[1])::NUMERIC;
$$;

CREATE AGGREGATE geo_mean (NUMBER)
(
    stype = NUMBER,
    initcond = '(0,0)', -- represent POINT value
    sfunc = sum_of_log,
    finalfunc = e_to_avg_of_log
);