clickhouse-aggregate-mvs-with-cdc.md

ClickHouse: Correct MV Aggregation Over CDC Data

Background

This document describes patterns for building materialized views over tables that use Change Data Capture (CDC) via ReplacingMergeTree. The core challenge is that MVs in ClickHouse are insert triggers, not recomputed views — so they see raw CDC rows before deduplication, and naive aggregations will be wrong.

CDC primer

When a row is superseded (updated or deleted), a new row is inserted with:

• The same ORDER BY key
• A higher _version
• _is_deleted = 1 if the row is being tombstoned

ReplacingMergeTree retains only the highest-_version row per key, but only after a merge — which is asynchronous. At any point before that merge, both the old and new rows coexist in the table.

Source table schema (generic)

CREATE TABLE events (
    id           UInt64,
    entity_id    UInt64,
    created_at   DateTime,
    response_id  UUID,
    x            Float64,
    _version     UInt64,  -- CDC version, monotonically increasing
    _is_deleted  UInt8    -- 1 = this row tombstones a prior version
)
ENGINE = ReplacingMergeTree(_version)
ORDER BY id;

---

Why Naive MVs Fail

MVs fire on each insert batch, before RMT merges.

When a backfill inserts a new version of a row, the MV fires again on that raw insert. It has no knowledge of what it previously wrote for that row. This causes:

• Double-counting for summable metrics — the new version adds to the sum again
• Stale distinct counts — cancelled response_ids remain in aggregate states with no way to remove them

The two patterns below solve this for the two most common metric types.

---

Pattern 1: Summable Metrics (sum(x), count(*))

Use a double-entry ledger approach. The MV records every insert as a signed ledger entry. CDC cancellations are recorded separately and subtracted at query time.

How it works

Every insert into events writes a row to the MV tagged with _is_deleted. At query time:

net = SUM(x) WHERE _is_deleted = 0
    - SUM(x) WHERE _is_deleted = 1

Because each CDC cancellation row carries the same x value as the row it supersedes, the two entries cancel exactly. A row that is inserted and then permanently deleted contributes x - x = 0. A row that is inserted, cancelled, and re-inserted contributes x - x + x = x.

Schema

CREATE TABLE daily_events_sum
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, _is_deleted)
AS
SELECT
    toDate(created_at) AS date,
    _is_deleted,
    sum(x)             AS x_sum
FROM events
GROUP BY date, _is_deleted;

Query

SELECT
    date,
    sumIf(x_sum, _is_deleted = 0)
  - sumIf(x_sum, _is_deleted = 1) AS net_x
FROM daily_events_sum
GROUP BY date
ORDER BY date;

No FINAL needed. SummingMergeTree merges ledger entries for the same key, and the subtraction resolves the net at query time.

Adding a business dimension

Any column can be added as an extra dimension — soft deletes, country, status, etc. For example, to exclude soft-deleted rows:

-- In the MV definition, add is_deleted to GROUP BY and ORDER BY
ORDER BY (date, is_deleted, _is_deleted)

-- At query time, filter the dimension you care about
SELECT
    date,
    sumIf(x_sum, _is_deleted = 0 AND is_deleted = 0)
  - sumIf(x_sum, _is_deleted = 1 AND is_deleted = 0) AS net_active_x
FROM daily_events_sum
GROUP BY date;

Critical invariant

The CDC cancellation row (_is_deleted = 1) must carry the exact same x value as the row it is cancelling.

If the cancellation row carries a different x, the subtraction will not cancel correctly. Enforce this in your backfill:

assert cancel_row['x'] == original_row['x'], (
    f"CDC cancel row for id={cancel_row['id']} has x={cancel_row['x']} "
    f"but original has x={original_row['x']}. These must match."
)

---

Pattern 2: Distinct Counts (count(distinct response_id))

Summable metrics are easy because +x - x = 0. Distinct counts are harder because aggregate states like uniqExact are append-only — once a response_id enters the state, it cannot be removed. Cardinality subtraction does not work because the same response_id legitimately appears in both live and cancelled rows.

The solution is to not aggregate at all in the MV. Instead, use ReplacingMergeTree on the MV target, keyed on (date, response_id). This stores one row per response_id per day, deduplicating by _version — exactly mirroring what RMT does on the source table, but scoped to the MV's grain.

Why (date, response_id) is a stable key

This only works if a given response_id always maps to the same date — i.e. its created_at timestamp never changes across CDC versions. If that holds, then (date, response_id) is a stable natural key and RMT will correctly collapse all versions of a row into the latest one.

Schema

CREATE TABLE daily_events_distinct
ENGINE = ReplacingMergeTree(_version)
PARTITION BY toYYYYMM(date)
ORDER BY (date, response_id)
AS
SELECT
    toDate(created_at) AS date,
    response_id,
    _version,
    _is_deleted
FROM events;

Query

SELECT
    date,
    countIf(_is_deleted = 0) AS active_response_count
FROM daily_events_distinct
FINAL  -- deduplicates by (date, response_id), keeps highest _version
GROUP BY date
ORDER BY date;

FINAL is required here, but it runs against the small pre-aggregated MV table rather than the large source table — making it cheap.

---

Correcting Stale MV Rows After a Backfill

Because MVs fire on raw inserts, a backfill that modifies existing rows (e.g. pausing a prompt, soft-deleting an entity) may leave stale rows in the MV target.

For Pattern 2 (RMT target), the backfill can correct this directly. Since the MV target is a regular table, the backfill can insert corrected rows into it with the latest _version. RMT will then prefer these over the stale rows.

-- Backfill inserts corrected rows directly into the MV target,
-- scoped to only the affected response_ids it already knows about
INSERT INTO daily_events_distinct
SELECT
    toDate(created_at) AS date,
    response_id,
    _version,
    _is_deleted
FROM events
FINAL
WHERE response_id IN (...)  -- affected keys, known to the backfill

This works because:

• The backfill queries events FINAL for only the affected keys — cheap targeted lookup
• The resulting rows carry the latest _version from the source, which is higher than whatever the MV wrote on the original insert
• RMT on the MV target deduplicates in favour of the corrected rows

For Pattern 1 (SMT target), this approach does not apply — SummingMergeTree accumulates rather than deduplicates. For summable metrics, the ledger pattern self-corrects as long as the CDC invariant (same x on cancel rows) holds.

---

Summary

Metric	MV engine	Needs FINAL at query time	Self-correcting via ledger	Backfill can patch MV directly
sum(x), count(*)	SummingMergeTree	No	Yes, if CDC invariant holds	No
count(distinct response_id)	ReplacingMergeTree	Yes (on small MV table)	No	Yes

When to skip MVs entirely

If correctness is more important than query speed, query the source table directly with FINAL:

SELECT
    toDate(created_at) AS date,
    countIf(_is_deleted = 0) AS active_response_count
FROM events
FINAL
GROUP BY date;

This is always correct and is a good baseline for auditing or reconciling MV output.