Time series and analytics: partition, roll up, scan columns

• OLTP: many short transactions, a few rows each, by key. Store rows.
• OLAP: few large queries, millions of rows, few columns. Store columns.
• Time series: append mostly, read by time window, recent data most.

A row store reads every column of the rows it scans. A column store reads only the columns the query names, and those compress well.

A full refresh recomputes every bucket of history. A rollup job writes only the last window.

I keep raw samples partitioned by day, and the dashboard reads rollups. A column store comes in when ad hoc scans outgrow Postgres.

CREATE TABLE metrics (
  ts    timestamptz      NOT NULL,
  host  int              NOT NULL,
  value double precision NOT NULL
) PARTITION BY RANGE (ts)1;

CREATE INDEX ON metrics (host, ts)2;

CREATE TABLE metrics_1m (
  minute timestamptz      NOT NULL,
  host   int              NOT NULL,
  n      bigint           NOT NULL,
  total  double precision NOT NULL4,
  lo     double precision NOT NULL,
  hi     double precision NOT NULL,
  PRIMARY KEY (minute, host)3
);

CREATE TABLE metrics_1h (
  hour  timestamptz      NOT NULL,
  host  int              NOT NULL,
  n     bigint           NOT NULL,
  total double precision NOT NULL,
  lo    double precision NOT NULL,
  hi    double precision NOT NULL,
  PRIMARY KEY (hour, host)
);

CREATE FUNCTION add_day_partition5(day date) RETURNS void LANGUAGE plpgsql AS $$
BEGIN
  EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF metrics FOR VALUES FROM (%L) TO (%L)',
    'metrics_' || to_char(day, 'YYYYMMDD'),
    day::timestamp AT TIME ZONE 'UTC',
    (day + 1)::timestamp AT TIME ZONE 'UTC');
END $$;

Raw samples go in one partition per day. Rollups keep count, sum, min and max per minute and per hour, so they combine.

I keep the OLTP primary for transactions. Changes flow from its WAL through a log into a column store, and analytics runs there.

Tool facts from each project's documentation: PostgreSQL, TimescaleDB, ClickHouse, BigQuery and Prometheus.

Postgres gives me partitions, pruning and upserts for rollups. A column store gives me compressed columns and aggregates on insert. Prometheus is for bounded host metrics.

every minute, for [now - 5 min, now):          // re-read a little, for late samples1
  group raw samples by (minute, host)
  upsert into the minute rollup: count, sum, min, max   // replace the row, never add to it2

every hour, for the last hour:
  group minute rows by (hour, host) into the hour rollup

dashboard: average = sum(total) / sum(n)              // never an average of averages3

every day:
  create the partition for tomorrow                   // a write never finds no partition
  drop the raw partition older than 30 days           // one file removed, no row scan4

INSERT INTO metrics_1m (minute, host, n, total, lo, hi)
SELECT date_bin('1 minute', ts, '2000-01-01 00:00+00'), host, count(*), sum(value), min(value), max(value)
FROM metrics
WHERE ts >= $1 AND ts < $2
GROUP BY 1, 2
ON CONFLICT (minute, host) DO UPDATE
SET n = EXCLUDED.n, total = EXCLUDED.total, lo = EXCLUDED.lo, hi = EXCLUDED.hi;

INSERT INTO metrics_1h (hour, host, n, total, lo, hi)
SELECT date_bin('1 hour', minute, '2000-01-01 00:00+00'), host, sum(n), sum(total), min(lo), max(hi)
FROM metrics_1m
WHERE minute >= $1 AND minute < $2
GROUP BY 1, 2
ON CONFLICT (hour, host) DO UPDATE
SET n = EXCLUDED.n, total = EXCLUDED.total, lo = EXCLUDED.lo, hi = EXCLUDED.hi;

CREATE FUNCTION add_day_partition(day date) RETURNS void LANGUAGE plpgsql AS $$
BEGIN
  EXECUTE format('CREATE TABLE IF NOT EXISTS %I PARTITION OF metrics FOR VALUES FROM (%L) TO (%L)',
    'metrics_' || to_char(day, 'YYYYMMDD'),
    day::timestamp AT TIME ZONE 'UTC',
    (day + 1)::timestamp AT TIME ZONE 'UTC');
END $$;

CREATE FUNCTION drop_day_partition(day date) RETURNS void LANGUAGE plpgsql AS $$
BEGIN
  EXECUTE format('DROP TABLE IF EXISTS %I', 'metrics_' || to_char(day, 'YYYYMMDD'));
END $$;

SELECT minute AS bucket, sum(n) AS n, sum(total) AS total, min(lo) AS lo, max(hi) AS hi
FROM metrics_1m
WHERE minute >= $1 AND minute < $2
GROUP BY 1 ORDER BY 1;

A job upserts count, sum, min and max per minute, then per hour. A daily job creates tomorrow's partition and drops the oldest.

On the minute rollup the 24-hour dashboard reads 9 times fewer rows. The hour rollup reads 360 times fewer for a week.

• The planner compares the filter with each partition's bounds and skips the rest.
• A prepared statement with a generic plan still prunes, when execution starts.
• Keep the count to a few thousand at most. Planning time and memory grow with it.

I filter on the partition column itself. A function around it hides the bounds, and the planner scans every partition.

The sample was off by 0.087%. SYSTEM picks whole pages, so rows that arrived together are sampled together.

For a total or an average I can read a sample. For distinct counts I use a HyperLogLog sketch.

encode(column, scheme):
  v = column
  IF delta:          v = each value minus the one before1
  IF delta-of-delta: v = delta applied twice2
  IF runs: write (value, count) per run of equal values3
  ELSE:    write each value as a varint4    // 1 byte: -64..63

  // every 10 s: 1000, 1010, 1020 ...  ->  1000, 10, 10 ...
  //   as runs:  (1000, 1) (10, 8639)5

// deltas returns the first value, then each value minus the one before it.
func deltas(xs []int64) []int64 {
  out := make([]int64, len(xs))
  for i, x := range xs {
    if i == 0 {
      out[i] = x
    } else {
      out[i] = x - xs[i-1]
    }
  }
  return out
}

// Encode writes a column with one scheme: a count, then the values. A regular timestamp column
// becomes all zeros after delta-of-delta, and run-length encoding stores those zeros as one pair.
func Encode(xs []int64, s Scheme) ([]byte, error) {
  out := binary.AppendUvarint(nil, uint64(len(xs)))
  vals := xs
  switch s {
  case Delta, DeltaRLE:
    vals = deltas(xs)
  case DeltaOfDelta, DeltaOfDeltaRL:
    vals = deltas(deltas(xs))
  }
  switch s {
  case Plain:
    for _, v := range vals {
      out = binary.LittleEndian.AppendUint64(out, uint64(v))
    }
  case Varint, Delta, DeltaOfDelta:
    for _, v := range vals {
      out = binary.AppendVarint(out, v) // zigzag: small negatives stay small
    }
  case RLE, DeltaRLE, DeltaOfDeltaRL:
    for i := 0; i < len(vals); {
      run := 1
      for i+run < len(vals) && vals[i+run] == vals[i] {
        run++
      }
      out = binary.AppendVarint(out, vals[i])
      out = binary.AppendUvarint(out, uint64(run))
      i += run
    }
  default:
    return nil, fmt.Errorf("unknown scheme %q", s)
  }
  return out, nil
}

• Sort the data by the columns you filter on: tenant or host, then time. Runs get long and values close.
• Pick an encoding per column, then add a general compressor such as LZ4 or ZSTD on top.
• A jittered timestamp has no long runs: plain delta wins at about 1 byte a value.

Sorted by host and time, a timestamp column is a few runs, and a gauge needs about 1.9 bytes a value. The three columns take 10.7 times less than rows.

Steps 1 to 3 are measured in the lab. Do not start at step 5: a warehouse adds a pipeline to run, and dashboards then lag the primary by seconds or minutes.

I start with one Postgres table and a time index. I add partitions, then rollups. A column store comes when ad hoc scans over all history matter.