create or replace function f (
  i_radius numeric,       --  压缩半径
  i_time timestamp,       --  开始时间
  i_interval_s numeric,   --  时间转换间隔 (秒，例如每5秒在坐标上表示1个单位间隔，则这里使用5) 
  query text,             --  需要进行旋转门压缩的数据, 例子 'select t, val from tbl where t>=%L order by t limit 100' ， select 子句必须固定， 必须按t排序
  OUT o_val numeric,      --  值，纵坐标 y  (跳跃点y)
  OUT o_time timestamp,   --  时间，横坐标 x (跳跃点x)
  OUT o_x numeric         --  跳跃点x, 通过 o_time 转换
)
returns setof record as $$
declare
  v_time timestamp;       -- 时间变量
  v_x numeric;            -- v_time 转换为v_x
  v_val numeric;          -- y坐标
  v1_time timestamp;      -- 前一点 时间变量
  v1_x numeric;           -- 前一点 v_time 转换为v_x
  v1_val numeric;         -- 前一点 y坐标
  v_start_time numeric;   -- 记录第一条的时间坐标, 用于计算x偏移量
  v_rownum int8 := 0;     -- 用于标记是否第一行
  v_max_angle1 numeric;   -- 最大上门夹角角度
  v_max_angle2 numeric;   -- 最大下门夹角角度
  v_angle1 numeric;       -- 上门夹角角度
  v_angle2 numeric;       -- 下门夹角角度
begin
  for v_time , v_val in execute format(query, i_time) 
  LOOP
    -- 第一行，第一个点，是实际要记录的点位
    v_rownum := v_rownum + 1;
    if v_rownum=1 then 
      v_start_time := extract(epoch from v_time);  
      v_x := 0;  
      o_val := v_val;  
      o_time := v_time;  
      o_x := v_x;  
      -- raise notice 'rownum=1 %, %', o_val,o_time;
      return next;  -- 返回第一个点  
    else
      v_x := (extract(epoch from v_time) - v_start_time) / i_interval_s;  -- 生成X坐标
      SELECT 180-ST_Azimuth(
                              ST_MakePoint(o_x, o_val+i_radius),    -- 门上点
                              ST_MakePoint(v_x, v_val)              -- next point
                           )/(2*pi())*360 as degAz,                 -- 上夹角
                 ST_Azimuth(
                              ST_MakePoint(o_x, o_val-i_radius),    -- 门下点
                              ST_MakePoint(v_x, v_val)              -- next point
                           )/(2*pi())*360 As degAzrev               -- 下夹角
      INTO v_angle1, v_angle2; 

      select GREATEST(v_angle1, v_max_angle1), GREATEST(v_angle2, v_max_angle2) into v_max_angle1, v_max_angle2;

      if (v_max_angle1 + v_max_angle2) >= 180 then  -- 找到四边形外的点位，输出上一个点，并从上一个点开始重新计算四边形
        -- raise notice 'max1 %, max2 %', v_max_angle1 , v_max_angle2;
        -- 复原
        v_angle1 := 0;
        v_max_angle1 := 0;
        v_angle2 := 0;
        v_max_angle2 := 0;

        -- 门已完全打开，输出前一个点的值
        o_val := v1_val; 
        o_time := v1_time; 
        v1_x := (extract(epoch from v1_time) - v_start_time) / i_interval_s;  -- 生成前一个点的X坐标 
        o_x := v1_x; 

        -- 用新的门，与当前点计算新的夹角 
        SELECT 180-ST_Azimuth(
                                ST_MakePoint(o_x, o_val+i_radius),    -- 门上点
                                ST_MakePoint(v_x, v_val)              -- next point
                             )/(2*pi())*360 as degAz,                 -- 上夹角
                   ST_Azimuth(
                                ST_MakePoint(o_x, o_val-i_radius),    -- 门下点
                                ST_MakePoint(v_x, v_val)              -- next point
                             )/(2*pi())*360 As degAzrev               -- 下夹角
        INTO v_angle1, v_angle2;

        select GREATEST(v_angle1, v_max_angle1), GREATEST(v_angle2, v_max_angle2) into v_max_angle1, v_max_angle2; 

        -- raise notice 'new max %, new max %', v_max_angle1 , v_max_angle2;

        -- raise notice 'rownum<>1 %, %', o_val, o_time;

        return next;
      end if; 

      -- 记录当前值，保存作为下一个点的前点
      v1_val := v_val; 
      v1_time := v_time; 
    end if; 
  END LOOP; 
end; 
$$ language plpgsql strict;

压缩测试

test=> 
select * from f (
  15, -- 门宽度=15
  '2016-08-12 23:22:27.530318', -- 开始时间
  1, -- 时间坐标换算间隔，1秒
  'select t, val from tbl where t>=%L order by t limit 100'  -- query
);

 o_val |           o_time           |       o_x        
-------+----------------------------+------------------
 18.23 | 2016-08-12 23:22:28.530443 |                0
  5.14 | 2016-08-12 23:22:29.530453 | 1.00001287460327
 90.25 | 2016-08-12 23:22:30.530459 | 2.00001883506775
......
 87.90 | 2016-08-12 23:24:01.53098  | 93.0005400180817
 29.94 | 2016-08-12 23:24:02.530985 | 94.0005450248718
 63.53 | 2016-08-12 23:24:03.53099  | 95.0005497932434
 12.25 | 2016-08-12 23:24:04.530996 | 96.0005559921265
 83.21 | 2016-08-12 23:24:05.531001 | 97.0005609989166
(71 rows)

test=> select val, t, (extract(epoch from t)-extract(epoch from first_value(t) over()))/1 as x from tbl where t>'2016-08-12 23:22:27.530318' order by t limit 100;
  val  |             t              |        x         
-------+----------------------------+------------------
 18.23 | 2016-08-12 23:22:28.530443 |                0
  5.14 | 2016-08-12 23:22:29.530453 | 1.00001001358032
 90.25 | 2016-08-12 23:22:30.530459 |  2.0000159740448
......
 83.21 | 2016-08-12 23:24:05.531001 | 97.0005581378937
 87.97 | 2016-08-12 23:24:06.531006 | 98.0005631446838
 58.97 | 2016-08-12 23:24:07.531012 | 99.0005691051483
(100 rows)

流式压缩的实现

本文略，其实也很简单，这个函数改一下，创建一个以数组为输入参数的函数。

以lambda的方式，实时的从流式输入的管道取数，并执行即可。

《HTAP数据库 PostgreSQL 场景与性能测试之 32 - (OLTP) 高吞吐数据进出(堆存、行扫、无需索引) - 阅后即焚(JSON + 函数流式计算)》

《HTAP数据库 PostgreSQL 场景与性能测试之 31 - (OLTP) 高吞吐数据进出(堆存、行扫、无需索引) - 阅后即焚(读写大吞吐并测)》

《HTAP数据库 PostgreSQL 场景与性能测试之 27 - (OLTP) 物联网 - FEED日志, 流式处理 与 阅后即焚 (CTE)》

《在PostgreSQL中实现update | delete limit - CTID扫描实践 (高效阅后即焚)》

方法1，阅后即焚：

流计算结果表（即有效位点），增加2个字段：PK于明细表关联，时间戳代表最后一条记录。下次阅后即焚从最后一个有效位点开始，从明细表继续消费。

方法2，直接在明细表上更新状态（点、当前记录是否可见）。

其他，所有涉及到中间计算结果的，都可以用类似方法实现：

计算当前记录时，更新计算结果到当前记录上（也就是通过直接更新明细表的流计算方法）。如果计算时需要用到上一条或者上若干条流计算结果，通过递归，或者使用UDF调用都很容易得到。

例子