An Aggregate Is a Function, Not a Sum

19 min read 192
An Aggregate Is a Function, Not a Sum

The Report That Counted Twice

The system I work on stores sales receipts. Every cash register sends every receipt, each receipt arrives through Kafka as a document, and ClickHouse keeps them.

On top of the raw documents sit the reports people actually look at: totals per cashier shift, totals per hour and per day for a store, revenue grouped by a custom field the merchant puts on the receipt.

Sounds like the textbook use case for ClickHouse. Insert rows, aggregate, done.

It took us four attempts.

The thing that makes it hard is not the volume. It is three facts about the stream that every naive design trips over:

time printed at the register printed a week earlier printed once delivered through Kafka aggregation running same receipt, delivered twice a week late pipeline down one line is one receipt: printed at the top, delivered at the bottom; a normal one takes seconds
Three things the stream does that a running sum cannot survive.
  • Documents come twice. Kafka delivery is at-least-once. A batch fails halfway, the consumer retries, the same receipt is inserted again. Any total that is "the sum of what was inserted" is now wrong.
  • Documents come late. Most receipts arrive within seconds, but a register that was offline for a week sends its backlog when it reconnects. A small but steady share of documents arrive more than a day late, and some a month late. A report for last Tuesday is never final.
  • The pipeline stops. Whatever computes the aggregates will sometimes be down: a deploy, a node restart, a bug. When it comes back, nothing may be lost.

And the reports are read with real latency budgets: an interactive call should answer in a few seconds, a heavy report over hundreds of registers in under thirty.

Four Attempts

#The ideaWhat broke
1Add it up on inserta duplicate is counted twice, forever
2Recompute from the wide tableread 46 million rows to use 508 thousand
3Remember every keythe memory of keys outgrew the node
4Don't store anything3.5 billion rows for one report

1. Add It Up on Insert

The first version was the classic one: materialized views on insert, writing partial sums into a SummingMergeTree. Each insert adds its contribution, background merges fold the rows together, and reports read pre-summed rows.

Kafka batch: receipt 1234, 500 consumer insert SummingMergeTree hour total: 500 batch fails halfway, retry Kafka the same batch again consumer insert again SummingMergeTree hour total: 1000
The second insert adds, it does not replace. Nothing in the total says which 500 came twice.

It is fast and it is wrong.

A re-delivered receipt adds its sum a second time, and once two numbers have been added there is no way to subtract the right one later. The aggregate has no memory of where its value came from.

2. Recompute From the Wide Table

The second version fixed the double counting with recomputation: a refreshable materialized view periodically rebuilt the totals of recently touched shifts and hours from the documents themselves, with duplicates collapsed by document key.

Correct this time. But it read the main documents table, which is wide (payload, statuses, dozens of columns) and sorted by register and document key, not by time.

The recompute filtered by date, and here is the part that bit us: in ClickHouse a date filter on a table not sorted by date cuts away whole partitions, but not granules (more on granules in a moment). One pass read 46 million rows to use 508 thousand:

older month skipped whole current month, sorted by register, then by document number register 1register 2 register 3register 4 register 5register 6 older to newer date filter 0 granules read 6 of 6 granules read whole, for the few newest rows in each
Schematic. The date cuts between months, but inside a month it cuts through every granule.

In production the passes started hitting the per-query memory limit and failed, again and again.

3. Remember Every Key

The third version went back to adding up on insert and tried to make duplicates impossible: before adding a receipt, check whether its key was seen before. The key lookup was implemented as a ClickHouse dictionary over a table of recent keys.

A dictionary lives in memory, and a key in it is not small: 161 bytes on our test cluster. A month of keys for every register did not fit, and from there it went in circles:

a month of keys too big for memory dictionary fails to load insert writes its rows, then fails consumer retries the same batch again and again receipt 1234 in storage copy 1 copy 2 copy 3 copy 4 and counting
Every retry wrote the rows before failing, so every retry stored one more copy.

Documents started landing several times each.

The fix for duplicates became a generator of duplicates.

4. Don't Store Anything

The fourth version gave up on stored aggregates altogether: compute every report from the documents at read time, collapsing duplicates with a GROUP BY on the document key. No state, nothing to get out of sync, re-deliveries and late documents are handled for free.

It was correct, and it was slow in a way no tuning could fix. A report over 600 shifts read about 3.5 billion rows. On our test cluster, at the 99th percentile:

Report1 client4 clients
600 shifts by key23.4 s119 s
hourly retail, 150 registers, a month17.0 sfailed on memory and time limits
daily retail, 150 registers, a month14.3 stimed out on every call

What the Failures Had in Common

Looking at the three production failures side by side was the turning point. None of them was caused by storing an aggregate:

stores anaggregate a sum can'tbe corrected readstoo much remembersevery key 1. add up on insert 2. recompute from the wide table 3. remember every key 4. store nothing what we wanted
Each failure sits in its own column, and none of them in the first. The last row is the design we were looking for.

Attempt 4 removed storage and paid for it on every read, which put it on the same failure as attempt 2.

So the question became: can an aggregate be stored, survive duplicates without remembering keys, and be recomputed without reading the world?

A Short Detour: Granules

To see why the read volume matters so much, it helps to know how ClickHouse reads.

A MergeTree table is split into parts, and each part is split into granules of 8192 rows by default. The primary index is sparse: it stores the first key of every granule, not of every row.

sparse primary index: the first key of every granule A, 0001 A, 5230 B, 0410 B, 8812 C, 0007 one part granule 18192 rows granule 2read whole granule 3read whole granule 4read whole granule 58192 rows WHERE register = 'B': granule 2 may end with B, so 2, 3 and 4 are read
The index knows where each granule starts, not what is inside it.

When a query filters by the primary key, ClickHouse finds the granules that may contain matching rows and reads them whole. Skip indexes (for example minmax over a date column) can throw away more granules, but only if the values inside a granule are close to each other.

The consequence: what a query reads depends far more on how rows are laid out than on how many rows it needs. If the rows you want are scattered across the table, you read the granules around each of them:

sorted by hour first: the rows we need are everywhere 8 of 8 granules read sorted by register first: the same rows sit together 2 of 8 granules read
The same twelve rows in both layouts. Only the sort order differs.

In our compute-at-read attempt, the narrow retail table was sorted by hour first. Within an hour, receipts of thousands of registers were interleaved, so a report for 150 registers touched nearly every granule of the month.

We even had to rewrite our own acceptance criteria: a limit on "rows read" was impossible to meet in any query shape, because the boundary granules always contain someone else's rows.

The good news is the reverse also holds. If the rows of one unit of work sit next to each other, reading them costs a handful of granules. That turned out to be the key to the final design.

The Idea: An Aggregate Is a Function

The design we kept rests on one shift in thinking. An aggregate is not the running sum of what was inserted. It is a function of its documents.

Take the smallest useful unit of aggregation, a cell:

case class CellKey(register: RegisterId, shift: Int, hour: Instant) // hour in UTC

Every document belongs to exactly one cell, and the link is not stored anywhere. It is computed from three columns of the document:

key(d) = ( register(d), shift(d), hour(t(d)) ) the cell a document belongs to; t(d) is the document's own time, hour() rounds it down to the hour D(k) = { d : key(d) = k } all documents of cell k, found by the rule, not by a stored link L(k) = { argmaxv ∈ D(k), id(v) = id(d) w(v) : d ∈ D(k) } for each document, its version with the latest write time w; a re-delivered document collapses here V(k) = Σd ∈ L(k) m(d) per measure; first and last numbers and dates take min and max

The same in code:

def key(d: Document): CellKey =
  CellKey(d.register, d.shift, d.date.truncatedTo(HOURS))

def docs(k: CellKey): Seq[Document] =
  allDocuments.filter(d => key(d) == k)

def value(k: CellKey): Measures =
  docs(k)
    .groupBy(_.id)
    .values
    .map(_.maxBy(_.writtenAt))        // a re-delivered document collapses here
    .foldLeft(Measures.empty)(_ + _)  // sums, counts, min, max

The value of a cell can be computed again at any moment, because the documents are still there and the rule is the same. A new computation is written as a new row, versioned by the time of recomputation, and replaces the old one in a ReplacingMergeTree.

This gives three properties for free:

  • A re-delivered document changes nothing. During recomputation the rows of one document are collapsed by document key (argMax by write time). Two deliveries become one row; the cell value stays the same. No dictionary, no key table, no memory of what was seen.
  • Recomputing twice is harmless. Recomputing a cell with no new documents produces the same value. On one node, two recomputations running at once both write a version, and the later one has seen at least as much, so it wins. Replicas are a different story (see What It Costs).
  • A late document is not lost. A receipt that is a month late marks its month-old cell as dirty now, and that cell is recomputed in the next pass.

The second half of the idea is additivity. Every measure in a cell is a sum, a count, a min or a max. Every document is in exactly one cell. So any report over any set of cells is just the combination of their values, without double counting:

11:0012:00 13:0014:00 register Bshift 12 70 110 90 130 register Ashift 75 120 80 200 150 hour 14:00 of the store 130 + 150 = 280 shift 75 of register A 120 + 80 + 200 + 150 = 550 day its 24 hours month its days sums and counts add up; min and max keep the first and last document numbers and dates
One grid, read in different directions. A week, a quarter or any set of stores is the same addition.

None of these needs a table of its own: a new report is a new query over the same cells. Even time zones work, as long as the offset from UTC is a whole number of hours, since a local day boundary is then still an hour boundary.

We checked the additivity the boring way: hours, days and shifts assembled from cells matched the same reports computed directly from documents on all 209 measures.

Not Only Receipts

Registers and receipts are just the example at hand. Swap the nouns and the pattern stays:

In this articleIn general
a registera source of events: a device, an account, a sensor
a receipt, stored as a documentan event with a stable key of its own
a shift, an hournatural windows the events fall into
sums, counts, min, maxmeasures that add up

It needs three things: every event carries everything needed to compute its cell key, that key does not change when the event is delivered again, and the measures add up, or at least merge. Where those hold, the aggregate can be a function of its events instead of a running sum.

Making It Work in ClickHouse

The idea is short. Most of the work was in the mechanics.

Kafka at least once consumer one insert per batch documents on the register's shard narrow documents by register, number dirty-cell journal which cells, since when written in one insert recompute, every 30 s cells register, shift, hour every 30 s shifts from cells reports read cells and shifts
The document, its narrow copy and its dirty mark are written by one insert. Everything after that is recomputation.

A Narrow Table With the Right Order

Recomputation never reads the wide documents table. A materialized view copies every document row into a narrow table with only the columns the measures need, sorted by (register, document number).

There is no time column in that key, and it does not need one. A register numbers its documents in the order it issues them, so the documents of one register in one hour sit next to each other in every part. Finding the documents of a cell is a short range scan. The densest cell in our test history, a little over 5 000 documents, reads 20 granules and finishes in 7 ms. Compared to reading the same minute of traffic from the wide table, the narrow one reads 30 to 37 times fewer rows.

A Journal Written in the Same Insert

A second materialized view on the same insert writes a dirty mark for every touched cell: the cell key and the time of the mark. No sums, just "this cell needs recomputing".

Both views run as part of the same insert as the document itself, so our code has no separate step between "document stored" and "cell marked". That is not a transaction, though. ClickHouse writes the source table and its views one after another, and if a view fails after the document is stored, the insert reports an error with the document already in place. Our consumer then retries the batch, which writes the mark again, and a repeated document is harmless here. How that retry meets insert deduplication on replicated tables, where a repeated block can be dropped before its views run, deserves a look of its own.

Status updates for documents arrive as separate rows without a date and shift, and the views simply skip them, so a status never triggers a recompute.

The journal is a ReplacingMergeTree keyed by the cell, so repeated marks of one cell collapse into one row with the latest time. Small inserts produced about 14 marks per cell per hour; after collapsing it was 1.16, and the main recompute pass became five times cheaper.

"Recomputed" Lives in the Data

A refreshable materialized view runs every 30 seconds and recomputes the cells that need it. The obvious way to track progress is a cursor, "processed up to time T". We did not keep one.

Instead, each pass compares two times taken from the data itself:

M(k) = max { t : (k, t) ∈ journal } the latest mark of cell k R(k) = max { t : (k, t) ∈ cells } the latest recomputation of cell k, or 0 if it was never computed dirty = { k : M(k) > R(k) − 60 s } marked after its last recomputation, with a minute of overlap pass = { (k, V(k), now) : k ∈ oldest(dirty, 50 000) } a new version per cell

The same in code:

def lastMark(k: CellKey): Instant =
  journal.marks(k).max

def lastRecompute(k: CellKey): Instant =
  cells.versions(k).maxOption.getOrElse(Instant.EPOCH) // never computed yet

def dirty(k: CellKey): Boolean =
  lastMark(k).isAfter(lastRecompute(k).minusSeconds(60))

def pass(): Seq[Cell] =
  journal.keys
    .filter(dirty)
    .sortBy(lastMark)
    .take(50_000)
    .map(k => Cell(k, value(k), recomputedAt = Instant.now()))

A cell that was never computed has no version yet, so it counts as computed at the epoch and turns dirty with its first mark.

A cursor has to be moved atomically together with the write, and it breaks under retries, parallel runs and restarts. The comparison does not care about any of that. After any outage, the next pass sees exactly the cells that were marked after their last recompute. The 60-second overlap covers the fact that parts of one insert become visible at slightly different moments, so a mark can appear just before its document does. The price is an occasional extra recompute.

A pass reads only the documents of the cells it recomputes.

Two Windows That Must Not Overlap

The main pass looks at marks from the last two hours. A second, catch-up view runs every ten minutes and looks at marks between two hours and one day old. It exists for outages longer than two hours.

The windows must not overlap. In an early version they did, and after an outage both views grabbed the same cells at the same time: 26 seconds and two times 2.8 GB of memory on one node for work that should have been done once.

Version by Snapshot Time

The version of a cell row is the time of its recomputation, not the latest write time of its documents. We tried the latter first. With many documents written in the same second, and recomputations running close together, the order of versions became undefined. A later snapshot always sees at least as many documents as an earlier one, so snapshot time is a version that is always correct.

That "at least as many" holds for snapshots taken on the same node. Across replicas it does not, and that is the weakest point of the whole design.

Shifts From Cells, Not From Documents

A shift is assembled from its cells by a second refreshable view: take the latest row of every cell of the shift, sum the measures, take min and max of numbers and dates. A busy register can have 142 000 documents in a shift, and some shifts stay open for months. Rebuilding a shift from documents on every new receipt would reread it all; from cells it is one row per hour of the shift.

Reading Without FINAL

Readers take the latest row of each key with LIMIT 1 BY over the key, ordered by recomputation time, and run that subquery on every shard over its local tables. Everything belonging to one register (documents, marks, cells, shifts) lives on the same shard, so "the latest row" is decided correctly locally and nobody waits for background merges.

Things That Bit Us

A few lessons that are not obvious from the design and cost us a day or more each.

Backfill Floods the Shift View

When we filled cells for 77 days of history in one go, every filled cell looked "recently recomputed" to the shift view. It then spent two hours regrouping millions of cells and writing nothing new.

first backfill backfill writes 77 days of cells recomputed_at = now shift view sees them as just recomputed regroups every cell 88M rows, 0.5 core a pass backfill, fixed backfill writes 77 days of cells recomputed_at = now - 2 days shift view sees them as two days old skips them 4M rows, almost no CPU
The shift view only looks at cells recomputed in the last 2 hours, its catch-up at the last day. Two days old is outside both; a new mark still gets a backfilled cell recomputed.

Backfilled rows now get a recomputation time two days in the past, outside both windows. The shift view went from 88 million rows and half a core per pass to 4 million rows and almost nothing.

Filter in a Subquery Inside Insert Views

On ClickHouse 26.3 we saw an insert view write column values shifted. Three things came together: output aliases that matched source column names, the filter in the same query, and blocks that mixed documents with status rows.

Moving the filter into a subquery fixed it, and we have a test for it:

-- values came out shifted
CREATE MATERIALIZED VIEW docs_mv TO docs AS
SELECT register_id, doc_number, ifNull(total, 0) AS total, ...
FROM transactions
WHERE type IS NOT NULL;

-- fixed
CREATE MATERIALIZED VIEW docs_mv TO docs AS
SELECT register_id, doc_number, ifNull(total, 0) AS total, ...
FROM (SELECT * FROM transactions WHERE type IS NOT NULL);

Granule Size Matters for Wide Rows

A cell has about 250 columns, and the retail report reads 189 of them. With the default granule of 8192 rows, a query for 150 registers decompressed about 1.1 GB to use 22 000 cells.

Smaller granules and smaller compressed blocks on the cell table:

SETTINGS index_granularity = 1024,
         min_compress_block_size = 8192,
         max_compress_block_size = 65536

The same query decompressed about 0.3 GB, and CPU dropped by about 30% for 9% more disk.

One Aggregate State per Document, Not per Column

Collapsing a document with a separate argMax per column builds dozens of aggregate states per document. One argMax over a tuple of the columns builds one:

-- one state per column
SELECT doc_key,
       argMax(total, updated_at),
       argMax(cash, updated_at),
       argMax(card, updated_at),
       ...
FROM docs GROUP BY doc_key;

-- one state per document
SELECT doc_key,
       argMax((total, cash, card, ...), updated_at) AS doc
FROM docs GROUP BY doc_key;

The Numbers

The test cluster is small on purpose: two shards on one machine, five cores per ClickHouse node, tens of thousands of registers, 77 days of history. Reports, 99th percentile:

cells computed from documents at read time 30 s report budget 1 client 4 clients 600 shifts by key hourly retail, 150 registers daily retail, 150 registers revenue by field, 50 registers 600 shifts by key, 1 client: cells 2.1 s2.1 600 shifts by key, 1 client: from documents 23.4 s23.4 hourly retail, 1 client: cells 2.5 s2.5 hourly retail, 1 client: from documents 17.0 s17.0 daily retail, 1 client: cells 1.3 s1.3 daily retail, 1 client: from documents 14.3 s14.3 revenue by field, 1 client: cells 0.6 s0.6 revenue by field, 1 client: from documents 7.9 s7.9 600 shifts by key, 4 clients: cells 2.7 s2.7 600 shifts by key, 4 clients: from documents 119 s119 hourly retail, 4 clients: cells 6.1 s6.1 hourly retail, 4 clients: from documents failed on memory and time limitsfailed daily retail, 4 clients: cells 3.7 s3.7 daily retail, 4 clients: from documents, every call timed out at 180 stimed out revenue by field, 4 clients: cells 1.6 s1.6 revenue by field, 4 clients: from documents 100 s100 1 s31030100 s 1 s31030100 s p99 response time, log scale; hollow marks: no answer at all
Every report from cells sits left of the budget line, under load as well. From documents, four clients were enough to push every report past it.
The same numbers as a table
ReportFrom documents, 1 clientCells, 1 clientFrom documents, 4 clientsCells, 4 clients
600 shifts by key23.4 s2.1 s119 s2.7 s
hourly retail, 150 registers, a month17.0 s2.5 sfailed6.1 s
daily retail, 150 registers, a month14.3 s1.3 stimed out3.7 s
revenue by custom field, 50 registers7.9 s0.6 s100 s1.6 s

The 600-shift report reads up to 360 thousand rows instead of 3.5 billion. Every report call on the cells stays within twice the theoretical granule bound for the rows it needs.

Correctness was checked against an independent reference: an expected value for every day, shift, hour and month of 1 000 registers, computed by the load generator without the service.

Zero mismatches across 31 checks, after three hours of live traffic in steps of increasing load, the last at three times the first.

What It Costs

Nothing is free, and these are the trade-offs we accepted on purpose.

handled on its own needs a hand no outage seen in 75-115 s down under 2 h main view catches up 2 h to 1 day catch-up view 1 to 7 days manual pass over 7 days marks have expired 2 h 1 day 7 days how long recomputation has been down
The windows bound how long an outage can be, not how late a document can be.
  • Lag. A report sees a new document one or two passes later, usually 75 to 115 seconds after insert. We decided not to compute the pending tail at read time.
  • Long outages need a hand. Marks older than a day are not picked up automatically. The journal keeps marks for seven days, so there is time to run a manual pass.
  • Replicas recompute on their own, and that is a hole. Refreshable views run on each replica of a shard, and each replica sees its own copy of the documents and of the journal, which replicate separately. A replica can see a mark before the document behind it, recompute the cell without that document, and stamp the result later than the correct version from the other replica. The later stamp wins. If the gap outlasts the 60-second overlap, nothing marks the cell again until its next document, and for a late document that may be never. Our test cluster had one replica per shard, so it never showed up there; we found this one on paper, while writing this post. The fixes we see: let one replica per shard recompute, derive the version from the documents rather than the clock, or sweep yesterday's cells once a day regardless of their stamps.
  • Only mergeable measures. Sums, counts, min and max add up across cells. Exact distinct counts and exact medians do not. ClickHouse can keep a mergeable aggregate state in a cell instead of a number: uniqState and quantileState combine across cells with uniqMerge and quantileMerge, approximately and in little space. The exact variants merge too, but carry every value. Our reports need none of this today; it is the direction the cells would grow in, not a hole in them.
  • Backfill is a procedure. Cells for existing history are built by a separate script, about an hour and a half for our 77 test days.

Closing

The four attempts were not four bad ideas. Each one fixed the previous failure and ran into the next property of the stream: duplicates, scattered reads, memory for deduplication, and finally the cost of reading everything on every request.

None of it is new, and we could have known that sooner. Databases call it incremental view maintenance: record which parts of a materialized result a change touched, recompute only those parts, replace the old values. Oracle's partition change tracking does exactly that for materialized views, and a survey from 1995 had already catalogued the techniques. It took us four attempts to reach the first page of that survey. What was ours is only the choice of the cell, and keeping the mark, the version and the "recomputed" flag inside ClickHouse tables themselves.

What finally worked was not a new ClickHouse feature. It was changing what an aggregate means. As long as an aggregate is a sum of inserts, every duplicate and every late document is a correction you have to remember. When it is a function of its documents, you never correct it. You just compute it again.