TimescaleDB连续聚合查询报错，如何修改使其生效？

问题背景 #

你编写了一段创建带timescaledb.continuous属性的物化视图的SQL代码，尝试将多个表的数据聚合为JSON结构，但执行时触发错误。

原SQL代码：

CREATE MATERIALIZED VIEW aggregate
WITH (timescaledb.continuous) AS
SELECT
json_build_object(
    'candles',
    (SELECT json_agg(array [EXTRACT(EPOCH FROM ts_bucket), "open", high, low, "close", volume :: BIGINT]) AS json_array_of_arrays
    FROM exchange.candles_d1
    WHERE exchange.candles_d1.ticker = 'BTCUSDT' AND candles_d1.ts_bucket >= time_bucket('1 hour', exchange.candles_d1.ts_bucket) AND candles_d1.ts_bucket < time_bucket('1 hour', exchange.candles_d1.ts_bucket) + INTERVAL '1 hour'),

    'kvwap',
    (SELECT json_agg(array [EXTRACT(EPOCH FROM ts_bucket), m1, m5, m15, m30, h1, h2, h4, d1, low, vwap, high]) AS json_array_of_arrays
    FROM exchange.kvwap_d1
    WHERE exchange.kvwap_d1.ticker = 'BTCUSDT' AND kvwap_d1.ts_bucket >= time_bucket('1 hour', exchange.candles_d1.ts_bucket) AND kvwap_d1.ts_bucket < time_bucket('1 hour', exchange.candles_d1.ts_bucket) + INTERVAL '1 hour'),

    'zones',
    (SELECT json_agg(array[EXTRACT(EPOCH FROM ts_confirmation)::BIGINT, EXTRACT(EPOCH FROM ts_end)::BIGINT, confirmations, CAST(is_continuation AS INT)]) AS json_array_of_array
    FROM analysis.zones
    WHERE analysis.zones.ticker = 'BTCUSDT' AND "interval" = 'D1' AND ts_confirmation >= time_bucket('1 hour', exchange.candles_d1.ts_bucket) AND analysis.zones.ts_confirmation < time_bucket('1 hour', exchange.candles_d1.ts_bucket) + INTERVAL '1 hour')
)
FROM exchange.candles_d1;

报错信息：

ERROR: invalid continuous aggregate query
Detail: CTEs, subqueries and set-returning functions are not supported by continuous aggregates.

你的疑问：确认json_agg是否不被连续聚合支持，同时寻求可行的解决办法。

问题原因 #

不是json_agg本身不被连续聚合支持，而是连续聚合不允许使用子查询、CTE或嵌套的聚合结构。你的SQL里每个JSON字段都嵌套了子查询，这直接违反了连续聚合的语法限制。

解决办法 #

采用「分层聚合」的思路：先为每个数据源创建单独的连续聚合（按小时粒度预聚合数据），再创建一个普通物化视图来组合这些预聚合的数据，生成最终的JSON结构。

步骤1：创建各表的连续聚合视图 #

1.1 针对candles_d1表

CREATE MATERIALIZED VIEW candles_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', ts_bucket) AS hour_bucket,
    json_agg(array [EXTRACT(EPOCH FROM ts_bucket), "open", high, low, "close", volume :: BIGINT]) AS candles_data
FROM exchange.candles_d1
WHERE ticker = 'BTCUSDT'
GROUP BY hour_bucket;

1.2 针对kvwap_d1表

CREATE MATERIALIZED VIEW kvwap_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', ts_bucket) AS hour_bucket,
    json_agg(array [EXTRACT(EPOCH FROM ts_bucket), m1, m5, m15, m30, h1, h2, h4, d1, low, vwap, high]) AS kvwap_data
FROM exchange.kvwap_d1
WHERE ticker = 'BTCUSDT'
GROUP BY hour_bucket;

1.3 针对zones表

CREATE MATERIALIZED VIEW zones_hourly
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', ts_confirmation) AS hour_bucket,
    json_agg(array[EXTRACT(EPOCH FROM ts_confirmation)::BIGINT, EXTRACT(EPOCH FROM ts_end)::BIGINT, confirmations, CAST(is_continuation AS INT)]) AS zones_data
FROM analysis.zones
WHERE ticker = 'BTCUSDT' AND "interval" = 'D1'
GROUP BY hour_bucket;

步骤2：创建普通物化视图组合数据 #

CREATE MATERIALIZED VIEW aggregate
AS
SELECT
    h.hour_bucket,
    json_build_object(
        'candles', COALESCE(c.candles_data, '[]'::json),
        'kvwap', COALESCE(k.kvwap_data, '[]'::json),
        'zones', COALESCE(z.zones_data, '[]'::json)
    ) AS aggregated_data
FROM (
    SELECT DISTINCT hour_bucket FROM candles_hourly
    UNION
    SELECT DISTINCT hour_bucket FROM kvwap_hourly
    UNION
    SELECT DISTINCT hour_bucket FROM zones_hourly
) h
LEFT JOIN candles_hourly c ON h.hour_bucket = c.hour_bucket
LEFT JOIN kvwap_hourly k ON h.hour_bucket = k.hour_bucket
LEFT JOIN zones_hourly z ON h.hour_bucket = z.hour_bucket;

补充说明 #

• 连续聚合负责按小时粒度预计算每个表的聚合结果，保证数据更新的高效性；
• 普通物化视图负责将多个预聚合结果组合成你需要的JSON结构，这里可以自由使用子查询、json_agg和json_build_object；
• 如果需要实时更新最终的aggregate视图，可以为它创建刷新触发器，或者定期执行REFRESH MATERIALIZED VIEW aggregate;。

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