基于LOCF的异构时间序列乘积连续聚合实现问询

我有两个采样时间不同的传感器，需要使用LOCF（最新值向前填充）计算小时时间桶内两个传感器值的乘积，按小时统计该乘积的时间加权结果，且需支持按客户ID查询数月或数年的小时级数据。

这看起来是TimescaleDB中continuous aggregate的适用场景，但我无法将现有查询转换为continuous aggregate。

传感器数据表结构 #

(
    "timestamp" timestamp with time zone NOT NULL,
    channel_id uuid NOT NULL, -- 关联传感器类型、客户ID等的传感器ID
    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-13:00时段）

注意：不能直接取小时平均值的乘积，这与乘积的小时时间加权结果完全不同。

我已经写出了能实现需求的SQL查询，但该查询使用了子查询、CTE和窗口函数，无法转换为continuous aggregate。请问是否有可行的实现方案？

现有实现SQL（已匿名处理） #

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' -- 实际条件更复杂，此处为示例
                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

核心思路：拆分连续聚合步骤，适配TimescaleDB的continuous aggregate限制 #

由于continuous aggregate对复杂窗口函数、多层CTE的支持有限，我们将计算拆分为多个简单的连续聚合步骤，逐步实现LOCF填充、乘积计算和时间加权统计：

步骤1：创建单传感器LOCF填充的连续聚合视图 #

为每个传感器单独创建细粒度的LOCF填充视图，利用last()函数天然支持的LOCF逻辑，规避复杂窗口函数：

-- 传感器A的分钟级LOCF连续聚合视图
CREATE MATERIALIZED VIEW sensor_a_locf_minutely
WITH (timescaledb.continuous) AS
SELECT
    time_bucket(INTERVAL '1 minute', r.timestamp) AS bucket,
    rc.customer_id,
    last(r.value, r.timestamp) AS s1_val -- last函数自动实现LOCF向前填充
FROM readings r
JOIN reading_channels rc ON rc.id = r.channel_id
WHERE rc.name = 's1-name'
GROUP BY bucket, rc.customer_id;

-- 传感器B的分钟级LOCF连续聚合视图
CREATE MATERIALIZED VIEW sensor_b_locf_minutely
WITH (timescaledb.continuous) AS
SELECT
    time_bucket(INTERVAL '1 minute', r.timestamp) AS bucket,
    rc.customer_id,
    last(r.value, r.timestamp) AS s2_val
FROM readings r
JOIN reading_channels rc ON rc.id = r.channel_id
WHERE rc.name = 's2-name'
GROUP BY bucket, rc.customer_id;

选择分钟桶是为了保证后续小时级聚合的时间切片精度，可根据实际采样频率调整桶粒度。

步骤2：创建小时级时间加权乘积的连续聚合视图 #

基于两个单传感器的LOCF视图，关联计算分钟级乘积，再聚合为小时级时间加权结果：

CREATE MATERIALIZED VIEW sensor_product_weighted_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket(INTERVAL '1 hour', a.bucket) AS hour_bucket,
    a.customer_id,
    time_weight('LOCF', a.bucket, a.s1_val * b.s2_val) AS weighted_product
FROM sensor_a_locf_minutely a
JOIN sensor_b_locf_minutely b 
    ON a.bucket = b.bucket 
    AND a.customer_id = b.customer_id
GROUP BY hour_bucket, a.customer_id;

步骤3：高效查询数据 #

按客户ID查询数月/数年数据时，直接查询最终的连续聚合视图即可，性能远优于原始复杂查询：

SELECT hour_bucket, weighted_product
FROM sensor_product_weighted_hourly
WHERE customer_id = 'joe_strommen'
AND hour_bucket BETWEEN '2024-01-01' AND '2024-06-01'
ORDER BY hour_bucket;

关键说明 #

1. 拆分逻辑的合理性：TimescaleDB continuous aggregate对简单聚合、单表操作支持更完善，拆分后每个步骤都符合其要求，避免了复杂嵌套逻辑的限制。
2. LOCF的简化实现：last()函数在连续聚合中会自动填充时间桶内的缺失值，等价于LOCF逻辑，无需手动用窗口函数分组处理。
3. 精度与性能平衡：细粒度分钟桶保证了时间加权计算的精度，而连续聚合会自动维护物化数据，查询数年数据时无需重复计算原始数据。
4. 数据刷新：可通过设置自动刷新策略（如ADD_MATERIALIZED_VIEW_POLICY）或手动执行REFRESH MATERIALIZED VIEW，保证视图数据的实时性。

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