我在使用TimescaleDB的 hypertable来存储传感器数据。需要帮助创建一个查询，使其能够作为连续聚合（Continuous Aggregate）工作，以便一次查询多年的数据。

数据来自两个传感器，它们在不同的时间发送数据。我需要在按小时为单位的时间区间中，使用LOCF（最近观测值向前填充）来计算这两个值的乘积。我需要对每个小时计算这个乘积，用户将按客户ID对按小时分段的数月或数年的数据进行查询。

这看起来是一个很合适的连续聚合用例，但我还没有弄清如何实现。

我把传感器数据存储在如下的表中：

(
    "timestamp" timestamp with time zone NOT NULL,
    channel_id uuid NOT NULL, -- ID of the sensor that maps to sensor type, customer ID, etc.
    value bigint NOT NULL,
)

例如，数据可能如下所示：

我对12:00的逐小时乘积的期望结果是

• 5 × 8 × (30 / 60)（表示12:00-12:30） +
• 7 × 8 × (10 / 60)（表示12:30-12:40） +
• 7 × 12 × (20 / 60)（表示12:40-1:00）

请注意，不能简单地把每小时的平均值相乘得到结果——这与把每小时的乘积再求平均的结果不同。

我已经能够写出一个SQL查询来实现我的目标。然而它使用子查询、CTE和窗口函数，因此不能做成连续聚合。

有没有办法实现呢？

下面是给出我期望结果的查询。由于我对具体的列名和表名进行了匿名处理，可能存在一些拼写错误。

SELECT
    time_bucket(INTERVAL '1 hour', timestamp),
    time_weight('LOCF', timestamp, s1s2_product)
FROM (
    SELECT
        timestamp,
        s1_val * s2_val AS s1s2_product
    FROM
    (
        WITH filled_data AS (
            SELECT 
                timestamp,
                s1_val,
                COUNT(s1_val) OVER (ORDER BY timestamp) AS s1_val_grp,
                s2_val,
                COUNT(s2_val) OVER (ORDER BY timestamp) AS s2_val_grp
            FROM (
                SELECT 
                    r.timestamp,
                    AVG(r.value) FILTER (WHERE rc.name = 's1-name') as s1_val,
                    AVG(r.value) FILTER (WHERE rc.name = 's2-name') as s2_val
                FROM readings r JOIN reading_channels rc ON rc.id = r.channel_id
                WHERE rc.name in ('s1-name', 's2-name')
                   AND rc.customer_id = 'joe_strommen' -- This actual condition is more complicated, but this is the general idea.
                GROUP BY r.timestamp
                ORDER BY r.timestamp
            )
        )
        SELECT 
            timestamp,
            FIRST_VALUE(s1_val) OVER (PARTITION BY s1_val_grp ORDER BY timestamp) AS s1_val,
            FIRST_VALUE(s2_val) OVER (PARTITION BY s2_val_grp ORDER BY timestamp) AS s2_val
        FROM filled_data
        ORDER BY timestamp
    )
)
GROUP BY time_bucket
ORDER BY time_bucket

解决方案

以下是测试数据

INSERT INTO "readings" ("timestamp", "channel_id", "value") VALUES
    ('2026-03-03 11:20:00', 'A', '5'),
    ('2026-03-03 11:50:00', 'B', '8'),
    ('2026-03-03 12:30:00', 'A', '7'),
    ('2026-03-03 12:40:00', 'B', '12'),
    ('2026-03-03 13:30:00', 'A', '8'),
    ('2026-03-03 13:40:00', 'B', '16'),
    ('2026-03-03 15:40:00', 'A', '10') ;

在 "reading" 表中的事件，包括用于报告的开始时间和结束时间（CTE "allPoints"）

        time1          |                         tsh                         
------------------------+-----------------------------------------------------
 2026-03-03 11:00:00+00 | ["..11:00:00+00","..12:00:00+00") -- start of report period
 2026-03-03 11:20:00+00 | ["..11:00:00+00","..12:00:00+00") -- change value for 'A'
 2026-03-03 11:50:00+00 | ["..11:00:00+00","..12:00:00+00") -- change value for 'B'
 2026-03-03 12:00:00+00 | ["..12:00:00+00","..13:00:00+00") -- start of hour 12:00:00
 2026-03-03 12:30:00+00 | ["..12:00:00+00","..13:00:00+00") -- change value for 'A'
 2026-03-03 12:40:00+00 | ["..12:00:00+00","..13:00:00+00") -- change value for 'B'
 2026-03-03 13:00:00+00 | ["..13:00:00+00","..14:00:00+00") -- end and start hour
 2026-03-03 13:30:00+00 | ["..13:00:00+00","..14:00:00+00") -- change value for 'A'
 2026-03-03 13:40:00+00 | ["..13:00:00+00","..14:00:00+00") -- change value for 'B'
 2026-03-03 14:00:00+00 | ["..14:00:00+00","..15:00:00+00") -- start of hour 14:00:00
(10 rows)

time1 - 事件的时间戳 - 某通道的新值，或小时的起始
tsH - 1小时时间范围

将通道的数值时间区间连接起来。

在聚合之前，我们得到如下数据

     tsuniqrange                   | lng |id1|channel1| v1 | id2 |channel2| v2 |       v |          tsh                         
-----------------------------------+-----+---+--------+----+-----+--------+----+---------+----------------------------------
 ["..11:00:00+00","..11:20:00+00") | 20.0|   |        |    |     |        |    |         | ["..11:00:00+00","..12:00:00+00")
 ["..11:20:00+00","..11:50:00+00") | 30.0| 1 | A      |  5 |     |        |    |         | ["..11:00:00+00","..12:00:00+00")
 ["..11:50:00+00","..12:00:00+00") | 10.0| 1 | A      |  5 |   2 | B      |  8 |   6.666 | ["..11:00:00+00","..12:00:00+00")
 ["..12:00:00+00","..12:30:00+00") | 30.0| 1 | A      |  5 |   2 | B      |  8 |  20.000 | ["..12:00:00+00","..13:00:00+00")
 ["..12:30:00+00","..12:40:00+00") | 10.0| 3 | A      |  7 |   2 | B      |  8 |   9.333 | ["..12:00:00+00","..13:00:00+00")
 ["..12:40:00+00","..13:00:00+00") | 20.0| 3 | A      |  7 |   4 | B      | 12 |  27.999 | ["..12:00:00+00","..13:00:00+00")
 ["..13:00:00+00","..13:30:00+00") | 30.0| 3 | A      |  7 |   4 | B      | 12 |  42.000 | ["..13:00:00+00","..14:00:00+00")
 ["..13:30:00+00","..13:40:00+00") | 10.0| 5 | A      |  8 |   4 | B      | 12 |  16.000 | ["..13:00:00+00","..14:00:00+00")
 ["..13:40:00+00","..14:00:00+00") | 20.0| 5 | A      |  8 |   6 | B      | 16 |  42.666 | ["..13:00:00+00","..14:00:00+00")
 ["..14:00:00+00","..15:00:00+00") | 60.0| 5 | A      |  8 |   6 | B      | 16 | 128.000 | ["..14:00:00+00","..15:00:00+00")
(10 rows)

id1, id2 - 来自表 "readings" 的行ID - 该周期的来源。
以及v1, v2 - 本周期内通道1、通道2 的取值

with 
,param as(
  select min(case when rc.name='name-A' then id end) channel_id1,
    min(case when rc.name='name-B' then id end) channel_id2
  from reading_channels rc 
  where  rc.customer_id='joe_strommen'
  group by rc.customer_id
)
,timerange as(
  select *,tstzrange(fromTime,toTime,'[]') tsH
  from(
    select cast('2026-03-03 11:00:00'  as timestamp) fromTime ,
           cast('2026-03-03 15:00:00'  as timestamp) toTime
  )a
)
,allPoints as(
  select distinct timestamp as time1  -- events in reading
    ,tstzrange(date_trunc('hour',timestamp),date_trunc('hour',timestamp)+interval '1 hour','[)') tsH
  from readings t 
  inner join timerange r on t.timestamp <@ tsH
  cross join param p
  where channel_id in(p.channel_id1,p.channel_id2)
  union 
  select hh time1  -- start, end of hours
    ,tstzrange(date_trunc('hour',hh),date_trunc('hour',hh)+interval '1 hour','[)')
  from timerange r 
  cross join lateral( select  hh from  generate_series(date_trunc('hour',fromTime)
           ,date_trunc('hour',toTime-interval '1 sec'),interval '1 hour')hh)
)
,allRanges as(
  select * 
    ,EXTRACT(EPOCH FROM (upper(tsUniqRange)-lower(tsUniqRange))) / 60 AS lng
  from(
    select 
      tstzrange(time1,(lead(time1,1,upper(tsH))
          over( order by time1)) ,'[)') tsUniqRange
      ,tsH
    from allPoints
    where time1 <@tsH
  )
  where tsUniqRange && tsH
)
,dataRanges as (
  select id,channel_id,timestamp as time1
    ,tstzrange(timestamp,(lead(timestamp,1,cast('9999-12-31 23:59:59' as timestamptz))
       over(partition by channel_id order by timestamp)) ,'[)') tsData
    ,value
  from readings t 
)
,rangeValues as(
select r.tsuniqrange,r.lng
  ,d1.id id1,d1.channel_id channel1,d1.value v1 
  ,d2.id id2,d2.channel_id channel2,d2.value v2 
  ,d1.value*d2.value*(lng/60) v
  ,tsH
from allRanges r
cross join param p
left join dataRanges d1 on d1.tsData && r.tsUniqRange and d1.channel_id=p.channel_id1 -- 'A'
left join dataRanges d2 on d2.tsData && r.tsUniqRange and d2.channel_id=p.channel_id2 -- 'B'
)
select tsH 
  ,sum(v1*v2*(lng/60.0)) resDraft
  ,sum(case when v1 is null and v1 is null then null
            when v1 is null then  1*v2*(lng/60.0)
            when v2 is null then  v1*1*(lng/60.0)
       else   v1*v2*(lng/60.0) 
       end) res
from rangeValues
group by tsH;

输出为

                         tsh                         |         resdraft         |           res            
-----------------------------------------------------+--------------------------+--------------------------
 ["2026-03-03 11:00:00+00","2026-03-03 12:00:00+00") |   6.66666666666666666680 |   9.16666666666666666680
 ["2026-03-03 12:00:00+00","2026-03-03 13:00:00+00") |  57.33333333333333333324 |  57.33333333333333333324
 ["2026-03-03 13:00:00+00","2026-03-03 14:00:00+00") | 100.66666666666666666656 | 100.66666666666666666656
 ["2026-03-03 14:00:00+00","2026-03-03 15:00:00+00") | 128.00000000000000000000 | 128.00000000000000000000
(4 rows)

在某一通道的数值未知时，我应用了两种计算方法。

如果表较大且出现性能问题，需要在CTE "dataRanges" 中使用WHERE条件来限制所选时间段，以便基于通道变化的频率来选取较短的区间。
如果通道值变化，例如每小时一次，可以通过将样本限定在“计费周期”前后各1 小时或1 天来实现。
或者另外计算该周期前后通道的首次数值变化。

Fiddle