Rollup aggregation

Deep dive

How it works

Parts:

• bucket rule — width and alignment, floor(ts / width) * width; day buckets depend on the time zone
• key — grouping dimensions kept (device, link id, customer); other columns are dropped
• partial — mergeable state per (bucket, key): count, sum, min, max (add sum of squares for variance)
• levels — raw, 1 min, 1 h, each with its own retention
• refresher — ingest-time fold, scheduled job, or incrementally maintained view
• sketch — mergeable state for percentiles (quantile sketches) and distinct counts (HyperLogLog)

Operations:

1. Fold: event (ts, key, v) goes to bucket ts / width; create the partial or merge Partial::one(v) into it.
2. Coarsen: merge fine partials into the coarser bucket. This equals aggregating the raw rows: count and sum add, min and max combine.
3. Read: mean = sum / count of the merged partial, never the mean of bucket means. Use the coarsest level that gives the resolution needed.
4. Late event: merge its one-row partial into its fine bucket and every coarser bucket above, or mark the bucket dirty and recompute.
5. Retention: each level expires on its own TTL, only after the next level is complete (time partitioning).

Where step 1 runs:

 raw (ts,key,v) --fold--> 1-min {count,sum,min,max} --merge 60:1--> 1-h {...} --merge 24:1--> 1-day
   keep N days            keep M weeks                  keep years               keep forever
 late event ------------> merge into its 1-min bucket and into every coarser bucket above it

Code

Std-only Rust: raw events to 1-minute to 1-hour partials, checked against direct aggregation.

use std::collections::BTreeMap;
use std::time::Instant;

/// Mergeable partial aggregate: merging two partials equals aggregating the union of their rows.
#[derive(Clone, Copy, Debug, PartialEq)]
struct Partial { count: u64, sum: i64, min: i64, max: i64 }

impl Partial {
    fn one(v: i64) -> Self { Partial { count: 1, sum: v, min: v, max: v } }
    fn merge(&mut self, o: &Partial) {
        self.count += o.count; self.sum += o.sum;
        self.min = self.min.min(o.min); self.max = self.max.max(o.max);
    }
    fn mean(&self) -> f64 { self.sum as f64 / self.count as f64 }
}

type Rollup = BTreeMap<(u64, u32), Partial>; // (bucket start in seconds, key) -> partial

/// Raw events (ts_s, key, value) -> one partial per (bucket, key).
fn from_raw(ev: &[(u64, u32, i64)], width: u64) -> Rollup {
    let mut r = Rollup::new();
    for &(t, k, v) in ev { add(&mut r, width, t, k, &Partial::one(v)); }
    r
}
fn add(r: &mut Rollup, width: u64, t: u64, k: u32, p: &Partial) {
    r.entry((t / width * width, k)).and_modify(|x| x.merge(p)).or_insert(*p);
}
/// Coarser level built from the finer level's partials only (no raw data).
fn coarsen(fine: &Rollup, width: u64) -> Rollup {
    let mut r = Rollup::new();
    for (&(t, k), p) in fine { add(&mut r, width, t, k, p); }
    r
}
fn median(v: &mut Vec<i64>) -> f64 { v.sort(); (v[(v.len() - 1) / 2] + v[v.len() / 2]) as f64 / 2.0 }

fn main() {
    let mut s = 0x9E3779B97F4A7C15u64; // xorshift64 for reproducible data
    let mut rnd = move || { s ^= s << 13; s ^= s >> 7; s ^= s << 17; s };
    // 24 h, 20 keys; busy minutes get more events and larger values, so counts are uneven
    let mut raw: Vec<(u64, u32, i64)> = (0..400_000).map(|_| {
        let t = (rnd() % 86_400) / 60 * 60 + rnd() % 60;
        let t = if rnd() % 4 == 0 { t / 3600 * 3600 + rnd() % 120 } else { t }; // burst at the start of each hour
        (t, (rnd() % 20) as u32, (t % 3600 / 60 + rnd() % 50) as i64)
    }).collect();
    let t0 = Instant::now();
    let min1 = from_raw(&raw, 60);
    let ns = t0.elapsed().as_nanos() as f64 / raw.len() as f64;
    let hour = coarsen(&min1, 3600);
    assert_eq!(hour, from_raw(&raw, 3600)); // merging partials == aggregating the raw rows, exactly
    println!("rows: raw {} | 1-min {} | 1-h {} | rollup {ns:.0} ns/event", raw.len(), min1.len(), hour.len());

    // average of averages is wrong when buckets hold different counts
    let (h, k) = (0u64, 3u32);
    let true_mean = hour[&(h, k)].mean();
    let means: Vec<f64> = min1.range((h, k)..(h + 3600, k)).filter(|(&(_, kk), _)| kk == k).map(|(_, p)| p.mean()).collect();
    let avg_of_avg = means.iter().sum::<f64>() / means.len() as f64;
    assert!((avg_of_avg - true_mean).abs() > 0.1);
    // a median cannot be merged from bucket medians (holistic); needs a sketch or the raw values
    let mut per_min: BTreeMap<u64, Vec<i64>> = BTreeMap::new();
    let mut all = Vec::new();
    for &(t, kk, v) in &raw { if kk == k && t < 3600 { per_min.entry(t / 60).or_default().push(v); all.push(v); } }
    let mut meds: Vec<i64> = per_min.values_mut().map(|v| median(v) as i64).collect();
    let (true_med, med_of_med) = (median(&mut all), median(&mut meds));
    assert_ne!(true_med, med_of_med);
    println!("key {k}, hour 0: mean {true_mean:.3} vs avg-of-avgs {avg_of_avg:.3}; median {true_med} vs median-of-medians {med_of_med}");

    // late event for a minute bucket that was already written: merge, do not rebuild
    let (mut min1, mut hour) = (min1, hour);
    let late = (125u64, k, 1_000i64);
    let p = Partial::one(late.2);
    assert!(min1.contains_key(&(120, k))); // that bucket already exists
    add(&mut min1, 60, late.0, k, &p);
    add(&mut hour, 3600, late.0, k, &p);
    raw.push(late);
    assert_eq!(min1, from_raw(&raw, 60));
    assert_eq!(hour, from_raw(&raw, 3600));
    assert_eq!(hour, coarsen(&min1, 3600));
    println!("late event merged; hour 0 of key {k}: count {}, max {}", hour[&(0, k)].count, hour[&(0, k)].max);
}

Output, measured on this Mac (Apple M4 Pro), release build: rows: raw 400000 | 1-min 28798 | 1-h 480, 66-76 ns per raw event for the BTreeMap fold (three runs). Key 3, hour 0: mean 45.792 against 52.516 (average of averages); median 44 against 49 (median of medians).

Complexity and constants

Fold: O(1) per event with a hash map, O(log n) with this BTreeMap. A partial is 32 bytes (four 8-byte fields) plus bucket and key. Rows ≤ buckets × active keys; in the run 13.9× fewer rows at 1 minute and 833× at 1 hour. Druid says rollup can cut row counts by orders of magnitude; the ratio depends on the workload.

Efficiency

• Cost: about 70 ns per raw event in the code above (measured on this Mac (Apple M4 Pro)). A query reads buckets in range × keys rows, not raw events. Bound: ingest CPU and memory for open buckets.
• Levers:
  • wider queryGranularity — more rows share a bucket (Druid docs: 5 minutes instead of 1) — when queries tolerate it
  • fewer, lower-cardinality dimensions, plus an abbreviated table beside the full one — better ratio (Druid docs) — most queries use few dimensions
  • hierarchical levels — each reads the one below (TimescaleDB docs) — long retention
  • skip raw storage — ClickHouse Null table engine feeding the view (ClickHouse docs) — raw rows not needed
• Naive → standard → state of the art:
  • aggregate raw rows on every query · O(raw rows in range)
  • partials per bucket, merged for any range · O(buckets in range)
  • tiered levels, invalidation tracking, sketches for holistic measures · O(changed buckets) per refresh

Implementation notes

• Keep partials as columns (count, sum, min, max) so the next level merges them (columnar file format); ClickHouse stores opaque aggregate states.
• Use integer sums (cents, microseconds) where exactness matters; float sums depend on merge order.
• Order by (key, bucket) for sequential per-key reads; the in-memory grouping is a hash aggregation.
• Write rollups while raw chunks are rewritten anyway: compaction or the chunk compactor.

Pitfalls

• Average of averages. Uneven counts make it wrong; the Prometheus rules guide says to aggregate numerator and denominator separately and never average a ratio or an average.
• Percentiles. Median of bucket medians is not the median (49 against 44 above). Store a quantile sketch per bucket; distinct counts need HyperLogLog.
• Late data after close. TimescaleDB real-time aggregates miss late rows in an already-materialized bucket until the next refresh; Druid streaming ingestion is best-effort, so several segments can hold rows that should have merged. Add a lateness margin (InfluxDB tasks take an offset) and a merge path.
• Lost dimensions. A dropped column cannot return; keep raw for N days or a wider second table.
• Cardinality explosion. Buckets × key combinations; a user id as dimension can leave one event per row.
• Day buckets. They depend on the zone; TimescaleDB continuous aggregates reject functions using the session time zone, so declare it in the view.
• Counter resets. Roll up rate() or increase() (Prometheus adjusts for resets), not raw counter values. Graphite downsamples by average unless set; totals need sum.
• Retention vs refresh. With an open-ended start_offset TimescaleDB drops aggregate rows when raw chunks are dropped; match the window to retention. Keep the incomplete bucket out (end_offset).

Variants and alternatives

In production

• TimescaleDB continuous aggregates: refresh policy (start_offset, end_offset, schedule_interval, default 24 h), aggregates on aggregates, real-time aggregates (off by default since v2.13).
• ClickHouse: AggregatingMergeTree replaces rows with the same sorting key by one row of aggregate states; insert with -State, read with -Merge. The materialized view runs on each inserted block. SummingMergeTree sums on merge, possibly incompletely, so queries still sum() and GROUP BY.
• Apache Druid: rollup on by default: rows with identical dimensions and timestamp after queryGranularity truncation become one. Rollup ratio = SUM(num_rows) / COUNT(*) with a count metric.
• Prometheus: recording rules precompute expressions into new series named level:metric:operations; raw retention defaults to 15 days.
• Graphite whisper: retentions = 15s:7d,1m:21d,15m:5y; a point is written to every archive at write time; xFilesFactor (default 0.5) and aggregationMethod (average default, also sum, min, max, last).
• InfluxDB 2 tasks: aggregateWindow(every: 1h, fn: mean) then to(bucket: ...).

Implementations

Sources

• J. Gray et al., Data Cube: A Relational Aggregation Operator Generalizing Group-By, Cross-Tab, and Sub-Totals, Data Mining and Knowledge Discovery 1997. paper · arXiv
• TimescaleDB docs: continuous aggregates, refresh policies, real-time aggregates, time zones
• ClickHouse docs: AggregatingMergeTree, SummingMergeTree, combinators, incremental materialized views
• Apache Druid: rollup
• Prometheus: recording rules, rule naming and aggregation, functions, storage
• Graphite: whisper, carbon configuration
• InfluxDB: downsample data with a task