如何基于动态条件计算滚动最小值？时序事件表实时qty最小值求解

我有如下数据表：

time        | pid  | qty | event 
---------------------+------+-----+-------
 2021-11-27 16:15:35 | 2207 |  1  | start  
 2021-11-27 16:15:12 | 2206 |  1  | stop   
 2021-11-27 16:00:11 | 2207 |  2  | stop   
 2021-11-27 15:51:43 | 2206 |  1  | start  
 2021-11-27 15:46:49 | 2206 |  4  | stop   
 2021-11-27 15:42:47 | 2206 |  4  | start  
 2021-11-27 15:41:36 | 2206 |  1  | stop   
 2021-11-27 15:41:29 | 2208 |  3  | start  
 2021-11-27 15:41:15 | 2207 |  2  | start  
 2021-11-27 15:39:58 | 2206 |  1  | start 

可通过以下SQL语句创建：

CREATE TABLE simple (
    time  TIMESTAMPTZ UNIQUE NOT NULL,
    pid   BIGINT,
    qty   BIGINT,
    event TEXT
);
INSERT INTO simple VALUES
    ('2021-11-27 16:15:35' , 2207 ,  1  , 'start'),
    ('2021-11-27 16:15:12' , 2207 ,  1  , 'stop '),  
    ('2021-11-27 16:00:11' , 2207 ,  2  , 'stop '),  
    ('2021-11-27 15:51:43' , 2206 ,  1  , 'start'),  
    ('2021-11-27 15:46:49' , 2206 ,  4  , 'stop '),  
    ('2021-11-27 15:42:47' , 2206 ,  4  , 'start'),  
    ('2021-11-27 15:41:36' , 2206 ,  1  , 'stop' ), 
    ('2021-11-27 15:41:29' , 2208 ,  3  , 'start'),  
    ('2021-11-27 15:41:15' , 2207 ,  2  , 'start'),  
    ('2021-11-27 15:39:58' , 2206 ,  1  , 'start');  

我需要针对每行对应的时间戳，计算截至该行所有处于生效状态（未触发stop）的start事件对应的最小qty值，预期输出结果如下：

time        | pid  | qty | event | min 
---------------------+------+-----+-------+-----
 2021-11-27 16:15:35 | 2207 |  1  | start | 1 -- 2207 min pid again
 2021-11-27 16:15:12 | 2206 |  1  | stop  | 3 -- 2208 min pid, only one not stopped
 2021-11-27 16:00:11 | 2207 |  2  | stop  | 1
 2021-11-27 15:51:43 | 2206 |  1  | start | 1 -- 2206 min pid again
 2021-11-27 15:46:49 | 2206 |  4  | stop  | 2
 2021-11-27 15:42:47 | 2206 |  4  | start | 2 
 2021-11-27 15:41:36 | 2206 |  1  | stop  | 2  -- 2206 stopped, now 2207 is min pid
 2021-11-27 15:41:29 | 2208 |  3  | start | 1  
 2021-11-27 15:41:15 | 2207 |  2  | start | 1  -- min pid is still 2206
 2021-11-27 15:39:58 | 2206 |  1  | start | 1  -- first

自定义聚合函数确实是该场景的最优方案，逐行累积维护状态的性能远高于关联子查询等实现，以下是PostgreSQL下的完整可运行方案：

步骤1：定义生效任务存储类型 #

用于存储当前所有未触发stop的start事件的pid和qty信息

CREATE TYPE active_task AS (pid BIGINT, qty BIGINT);

步骤2：定义状态转换函数 #

逐行处理数据更新生效任务列表：遇到start事件就新增记录，遇到stop事件就删除同pid同qty的最早匹配记录

CREATE OR REPLACE FUNCTION update_active_tasks(state active_task[], row_time TIMESTAMPTZ, row_pid BIGINT, row_qty BIGINT, row_event TEXT)
RETURNS active_task[] AS $$
BEGIN
    IF trim(row_event) = 'start' THEN
        state := state || (row_pid, row_qty)::active_task;
    ELSIF trim(row_event) = 'stop' THEN
        FOR i IN 1..array_length(state, 1) LOOP
            IF state[i].pid = row_pid AND state[i].qty = row_qty THEN
                state := array_remove(state, state[i]);
                EXIT;
            END IF;
        END LOOP;
    END IF;
    RETURN state;
END;
$$ LANGUAGE plpgsql IMMUTABLE;

步骤3：定义结果计算函数 #

从当前生效任务列表中提取最小qty值

CREATE OR REPLACE FUNCTION get_min_active_qty(state active_task[])
RETURNS BIGINT AS $$
BEGIN
    RETURN (SELECT min(qty) FROM unnest(state) AS t(qty));
END;
$$ LANGUAGE plpgsql IMMUTABLE;

步骤4：创建自定义聚合函数 #

CREATE AGGREGATE min_active_qty(TIMESTAMPTZ, BIGINT, BIGINT, TEXT) (
    SFUNC = update_active_tasks,
    STYPE = active_task[],
    FINALFUNC = get_min_active_qty,
    INITCOND = '{}'
);

步骤5：执行查询 #

按时间升序累积计算，最终按时间降序输出即可得到和预期完全一致的结果

SELECT 
    time, pid, qty, trim(event) AS event,
    min_active_qty(time, pid, qty, event) OVER (ORDER BY time ASC) AS min
FROM simple
ORDER BY time DESC;

内容的提问来源于stack exchange，提问作者blunderedbus