Downsampling with TimescaleDB Continuous Aggregates #

A dashboard that scans four million raw readings to draw a week of hourly means is doing a thousand times more work than the answer requires, every time someone loads the page. Continuous aggregates fix that by materializing the buckets incrementally as data arrives — but the naive definition breaks the moment a reading turns up late, silently serving a bucket that was correct when it was computed and is not any more. This guide writes aggregates that stay correct, refresh policies that absorb late data, and the real-time view that keeps the newest hour visible. It implements the rollup layer of time-series storage partitioning and retention.

What a Continuous Aggregate Actually Is #

It is a materialized view that TimescaleDB maintains incrementally. Rather than recomputing from scratch, it tracks which time buckets have changed since the last refresh and re-materializes only those. Two consequences follow, and both matter.

First, refresh cost is proportional to the data that changed, not to the table size — which is why an aggregate over three billion rows refreshes in seconds.

Second, correctness depends entirely on the refresh window. A bucket is re-materialized only if the policy’s range covers it. If your policy refreshes the last hour and a reading arrives four days late, the bucket it belongs to is never recomputed and the aggregate is permanently wrong for that hour. Nothing errors; the number is simply stale forever.

Production-Ready Implementation #

The definition first. Note what is carried alongside the mean:

-- postgresql 15 · timescaledb 2.14 · postgis 3.4
CREATE MATERIALIZED VIEW reading_hourly
WITH (timescaledb.continuous) AS
SELECT device_id,
       metric,
       time_bucket('1 hour', observed_at)                      AS bucket,
       avg(value)                                              AS mean_value,
       min(value)                                              AS min_value,
       max(value)                                              AS max_value,
       count(*)                                                AS n_total,
       count(*) FILTER (WHERE qc_flag IN (1, 2, 3))            AS n_measured,
       max(qc_flag)                                            AS worst_flag,
       sqrt(sum(uncertainty * uncertainty)) / nullif(count(*), 0) AS mean_uncertainty
FROM   reading
GROUP  BY device_id, metric, bucket
WITH NO DATA;

CREATE INDEX ON reading_hourly (device_id, bucket DESC);

WITH NO DATA avoids materializing years of history inside the migration that creates the view — backfill it afterwards in bounded chunks, or the DDL will hold locks for hours.

Two aggregate expressions deserve comment. max(qc_flag) propagates the worst flag in the bucket, which works because the flag vocabulary is severity-ordered. And the uncertainty expression combines in quadrature and divides by n, which is the correct propagation for independent errors — the systematic component has to be added back by the consumer, as uncertainty propagation explains.

The refresh policy is where late data is handled:

SELECT add_continuous_aggregate_policy('reading_hourly',
    start_offset      => INTERVAL '30 days',   -- re-check buckets this far back
    end_offset        => INTERVAL '1 hour',    -- leave the newest hour to the real-time view
    schedule_interval => INTERVAL '15 minutes');

start_offset must exceed your worst late-arrival latency. For a network with store-and-forward gateways that can be offline for a week, thirty days is a sensible margin; for an all-cellular network, three days is ample. The cost of a generous window is bounded — only buckets whose rows actually changed are recomputed — so err large.

Daily rollups build on the hourly aggregate rather than on raw data:

CREATE MATERIALIZED VIEW reading_daily
WITH (timescaledb.continuous) AS
SELECT device_id,
       metric,
       time_bucket('1 day', bucket)                    AS day,
       avg(mean_value)                                 AS mean_value,
       max(max_value)                                  AS max_value,
       sum(n_total)                                    AS n_total,
       sum(n_measured)                                 AS n_measured,
       count(*) FILTER (WHERE n_measured >= 45)        AS valid_hours   -- 75% of 60
FROM   reading_hourly
GROUP  BY device_id, metric, day
WITH NO DATA;

valid_hours encodes the data-capture rule at the daily level, so a consumer can require, say, 18 valid hours before treating a daily mean as reportable — without re-deriving the rule in every query.

Backfilling in chunks keeps the initial materialization from locking the table:

# python 3.11 · psycopg[binary]==3.1.18
import psycopg


def backfill(conn: psycopg.Connection, view: str, start: str, end: str,
             chunk_days: int = 30) -> None:
    """Materialize history a month at a time. One call over three years will
    hold locks long enough to stall ingest; chunking keeps each transaction short."""
    with conn.cursor() as cur:
        cur.execute(
            "SELECT generate_series(%s::timestamptz, %s::timestamptz, %s::interval)",
            (start, end, f"{chunk_days} days"),
        )
        windows = [row[0] for row in cur.fetchall()]

    for lo, hi in zip(windows, windows[1:]):
        with conn.cursor() as cur:
            cur.execute("CALL refresh_continuous_aggregate(%s, %s, %s)", (view, lo, hi))
        conn.commit()

Parameter Tuning Guide #

Verification and Testing #

The test that matters proves late data is absorbed rather than lost.

-- 1. A bucket materialized before its late reading arrives
INSERT INTO reading VALUES ('s1', '2026-08-01T10:05Z', 'pm25', 20.0, 1);
CALL refresh_continuous_aggregate('reading_hourly', '2026-08-01Z', '2026-08-02Z');

SELECT mean_value, n_total FROM reading_hourly
WHERE device_id = 's1' AND bucket = '2026-08-01T10:00Z';   -- 20.0, 1

-- 2. The late reading lands in the already-materialized bucket
INSERT INTO reading VALUES ('s1', '2026-08-01T10:40Z', 'pm25', 30.0, 1);

SELECT mean_value, n_total FROM reading_hourly
WHERE device_id = 's1' AND bucket = '2026-08-01T10:00Z';   -- STILL 20.0, 1 — stale

-- 3. After a refresh covering that bucket, it is correct
CALL refresh_continuous_aggregate('reading_hourly', '2026-08-01Z', '2026-08-02Z');

SELECT mean_value, n_total FROM reading_hourly
WHERE device_id = 's1' AND bucket = '2026-08-01T10:00Z';   -- 25.0, 2

Step 2 is the entire lesson: between the late insert and the next refresh, the aggregate serves a number that is wrong and looks fine. Everything about the refresh policy exists to bound how long that window lasts.

Alongside it, run an agreement check on a sample of buckets:

def test_aggregate_matches_raw_for_a_sample_of_buckets(conn):
    """Any disagreement means a refresh window is too narrow for the real late-data tail."""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT a.device_id, a.bucket, a.mean_value, r.mean_value
            FROM   reading_hourly a
            JOIN   LATERAL (
                     SELECT avg(value) AS mean_value FROM reading
                     WHERE device_id = a.device_id AND metric = a.metric
                       AND observed_at >= a.bucket AND observed_at < a.bucket + interval '1 hour'
                   ) r ON true
            WHERE  a.bucket > now() - interval '7 days'
            ORDER  BY random() LIMIT 200
        """)
        for device_id, bucket, agg, raw in cur.fetchall():
            assert abs(agg - raw) < 1e-9, f"{device_id} {bucket}: {agg} vs {raw}"

Gotchas #

end_offset of zero. It materializes the currently-open bucket, which is incomplete by definition and will change with every subsequent reading. Leave at least one bucket unmaterialized and let the real-time view serve it.

Refreshing only the last hour. The most common misconfiguration, and the one that produces permanently stale buckets for every late reading. Size start_offset from your measured arrival latency.

Aggregating an aggregate without weighting. avg(mean_value) over hourly buckets is the mean of means, which equals the true mean only when every bucket has the same count. For daily figures where coverage varies, weight by n_total or compute from raw.

Compressing chunks that the aggregate still refreshes. A compressed chunk is effectively read-only; a refresh that needs to re-materialize a bucket inside one will fail. Set the compression policy’s offset beyond the aggregate’s start_offset.

FAQ #

Related #

• Time-Series Storage Partitioning and Retention — the storage stage these aggregates belong to
• Designing a Retention Policy for High-Frequency Telemetry — why aggregates outlive the raw data they were computed from
• Implementing Tumbling Windows for Air Quality Metrics — the streaming equivalent of the same bucketing