ClickHouse Stream Analytics: Duplicates, Late Events, and Replay

A purchase stream rarely arrives in a clean sequence. A client retries after a timeout, a worker delivers the same event again, and a delayed event appears after yesterday’s report has already been viewed.

To make the problem concrete, start with 100 distinct purchase events. Ten are delivered twice, and five of the distinct events arrive the following day.

The completed report should still describe 100 purchases. Replaying the same logical events should preserve that answer.

ClickHouse stream analytics needs an explicit design for event identity, reporting time, and the way derived results respond to repeated or delayed input. Fast aggregation becomes useful when there is also a known answer against which to check it.

This article develops a count-only example with immutable purchase events. Refunds, corrections, and revenue require additional contracts; a distinct purchase count does not establish those results automatically.

An event describes one business fact. A delivery describes one attempt to transport that fact. Preserve the event identity when delivery is retried or replayed.

For this example, a purchase receives one stable event_id when it is confirmed. The ID is unique within its tenant and source. All deliveries of that ID preserve its event time and meaning. Separate purchases receive separate IDs, even when their timestamps and amounts match.

Reusing an ID for a different purchase silently combines two facts. Changing an ID during retry silently splits one fact into two. If producers can disagree, define which producer owns canonical identity and how conflicting payloads are detected before they enter the analytical dataset. Purchase Event Deduplication explains the browser, server, and retry paths behind that problem.

Here is a small raw table for an isolated example database:

CREATE TABLE purchase_events
(
  tenant_id String,
  source_id String,
  event_id String,
  delivery_id UUID,
  event_time DateTime64(3, 'UTC'),
  ingested_at DateTime64(3, 'UTC')
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(event_time)
ORDER BY (tenant_id, source_id, event_time, event_id);

The sorting key supports scoped time queries. It does not enforce uniqueness for event_id. This raw table deliberately allows repeated deliveries so the example can inspect both transport behavior and logical purchase counts.

In a production ingestion path, also define schema version, validation, and quarantine behavior. A payload that reuses an ID with a different reporting day violates this article’s immutable-event contract. Merely applying uniqExact independently within each day would count it in both days. Event Schema Versioning develops the compatibility rules that keep the event’s meaning stable.

Start with a direct query whose meaning is easy to explain. Count raw rows as deliveries and distinct event IDs as purchases, within tenant, source, and UTC day:

SELECT
  tenant_id,
  source_id,
  toDate(event_time, 'UTC') AS day,
  count() AS deliveries,
  uniqExact(event_id) AS purchases
FROM purchase_events
GROUP BY tenant_id, source_id, day
ORDER BY tenant_id, source_id, day;

This is a reference for the count defined above, assuming the identity and immutability rules hold. An authenticated production report should restrict the tenant, authorized source, and reporting period before aggregation. The unrestricted query is useful in an isolated fixture database.

uniqExact provides an exact distinct count. Its state grows with the number of unique values, so accuracy has a memory cost. Keep this reference for known fixtures and bounded validation periods; benchmark the production cardinality before choosing it for every report. The function documentation explains that trade-off.

Generate 95 initial events for October 7. The SQL examples use synchronous inserts to keep the fixture stages easy to observe:

INSERT INTO purchase_events
SELECT
  'tenant-a',
  'store-1',
  concat('purchase-', toString(number + 1)),
  generateUUIDv4(),
  addMinutes(toDateTime64('2026-10-07 09:00:00', 3, 'UTC'), number),
  addMinutes(toDateTime64('2026-10-07 09:00:05', 3, 'UTC'), number)
FROM numbers(95)
SETTINGS async_insert = 0;

Deliver the first ten logical events again. Their event IDs and event times stay the same; their delivery IDs and ingestion times change:

INSERT INTO purchase_events
SELECT
  'tenant-a',
  'store-1',
  concat('purchase-', toString(number + 1)),
  generateUUIDv4(),
  addMinutes(toDateTime64('2026-10-07 09:00:00', 3, 'UTC'), number),
  addMinutes(toDateTime64('2026-10-07 12:00:00', 3, 'UTC'), number)
FROM numbers(10)
SETTINGS async_insert = 0;

The reference now returns 105 deliveries and 95 purchases. This distinction is the first useful invariant: duplicate transport input can increase the delivery log without increasing the business count.

ClickHouse can recognize certain repeated inserts and avoid storing them again. That protection depends on the table engine, query settings, retained deduplication history, and the identity of the inserted blocks or supplied token. Its window is finite. Insert deduplication on retries documents those conditions.

Treat that as protection for a retrying insertion, alongside the application’s event identity contract. A logical event may be delivered in a different batch, much later, or with new delivery metadata. A long-term replay can also outlive the retained insert history. It still needs to produce the same logical answer.

In this fixture, repeated deliveries contain a new delivery_id, so they are deliberately different rows. We want those deliveries visible in the raw log. Production systems can instead keep transport attempts in a separate log and deduplicate the analytical input, provided the resulting guarantees remain explicit.

If a writer uses an insertion token, reuse it for retries of that insertion operation and distinguish it from another operation’s token. It identifies an insert, not every business event contained in arbitrary future batches. Record the server version and effective settings when validating this behavior; defaults and asynchronous insertion behavior can differ across versions and compatibility modes.

Test two separate questions: does retrying a particular insert avoid duplicate storage under the configured mechanism, and does the business aggregate remain correct when duplicate rows nevertheless reach it? Passing one does not establish the other.

The latter question also exposes an easy mistake with replacement engines. Removing repeated raw rows during later merges cannot repair an earlier additive count simply because the source eventually looks deduplicated. The derived count needs a design that preserves identity itself or a rebuild path.

An incremental materialized view processes newly inserted blocks and writes its result into a target table. It does not continuously rerun the full raw query. The incremental materialized view documentation describes this insert-driven behavior.

A target that adds count() from each block counts repeated deliveries. Even adding a distinct count calculated separately inside each block is insufficient: an ID present in two blocks contributes once to each partial count.

For this immutable count example, preserve the distinct state across blocks. uniqExactState records the set state; uniqExactMerge combines those states before producing the count. ClickHouse’s aggregate combinators and AggregatingMergeTree describe this pattern.

Create the target and view before running the fixture inserts in a fresh database:

CREATE TABLE purchase_daily
(
  tenant_id String,
  source_id String,
  day Date,
  delivery_state AggregateFunction(count),
  purchase_state AggregateFunction(uniqExact, String)
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(day)
ORDER BY (tenant_id, source_id, day);

CREATE MATERIALIZED VIEW purchase_daily_mv
TO purchase_daily
AS
SELECT
  tenant_id,
  source_id,
  toDate(event_time, 'UTC') AS day,
  countState() AS delivery_state,
  uniqExactState(event_id) AS purchase_state
FROM purchase_events
GROUP BY tenant_id, source_id, day;

Read the target by merging states across all stored parts for each key:

SELECT
  tenant_id,
  source_id,
  day,
  countMerge(delivery_state) AS deliveries,
  uniqExactMerge(purchase_state) AS purchases
FROM purchase_daily
GROUP BY tenant_id, source_id, day
ORDER BY tenant_id, source_id, day;

The query supplies the aggregation semantics even before background merges combine target rows. Counting physical target rows or reading individual partial counts would answer a different question.

At every fixture stage, compare all tenant/source/day keys and both measures with the raw reference. Add another tenant and source to challenge grouping boundaries. A global total can remain correct while records are assigned to the wrong owner or day.

This design preserves the purchase count under repeated immutable input. It does not remove repeated raw rows, deduplicate a revenue sum, or retract a purchase that was later corrected. Those operations need their own state and reconciliation rules. Keep the exact-state memory cost visible in load tests too.

Now insert purchases 96 through 100. They occurred on October 7 and entered ingestion on October 8:

INSERT INTO purchase_events
SELECT
  'tenant-a',
  'store-1',
  concat('purchase-', toString(number + 1)),
  generateUUIDv4(),
  addMinutes(toDateTime64('2026-10-07 09:00:00', 3, 'UTC'), number),
  addMinutes(toDateTime64('2026-10-08 01:00:00', 3, 'UTC'), number - 95)
FROM numbers(95, 5)
SETTINGS async_insert = 0;

October 7’s purchase count becomes 100. The five deliveries arrived on October 8, but the report groups by event time. An operational ingestion chart could group by ingested_at and correctly show those arrivals on October 8. Both charts need labels that explain their clocks.

These are expected results for this dataset, not measured production metrics.

The reporting period also needs a completion policy. A “last updated” timestamp reports freshness; it does not prove that every event has arrived. A source checkpoint or documented lateness policy can support a provisional or finalized state, but a future backfill may still require reopening a period.

For an illustrative policy, show recent periods as provisional, display when they were recalculated, and identify revisions caused by later input. If the business requires an immutable published report, store a versioned report snapshot and publish a correction separately. Choose the behavior according to what readers are expected to rely on.

Invalidate the cache for the affected event-time period when late events arrive. Invalidating only today’s ingestion-day cache leaves yesterday’s report stale even though the underlying aggregate has changed. Designing real-time analytics with ClickHouse connects this reporting contract to freshness and query design.

A replay delivers the same logical facts again. It should preserve the stable event IDs, tenant/source ownership, and event times while making the new delivery attempts distinguishable.

For this fixture, replay each distinct event once from the raw table:

INSERT INTO purchase_events
SELECT
  tenant_id,
  source_id,
  event_id,
  generateUUIDv4(),
  event_time,
  toDateTime64('2026-10-09 10:00:00', 3, 'UTC')
FROM purchase_events
GROUP BY tenant_id, source_id, event_id, event_time
SETTINGS async_insert = 0;

This adds 100 deliveries to the previous 110. The raw reference and materialized result should both return 210 deliveries and 100 purchases for October 7. Replay them again with different batch sizes and ordering. The delivery count should increase by the number of accepted rows; the purchase count should remain 100.

Keep the expected result independent from the query implementation. Generate a manifest of the 100 known IDs and their tenant, source, and day before ingestion. Compare the raw query with that manifest, then compare the aggregate with the raw query. Otherwise, the same grouping mistake in two SQL queries can make them agree on the wrong answer.

Rebuilding an aggregate is a related but separate operation. Creating an incremental view after data already exists does not automatically populate all historical state. A backfill must account for those rows and for new live input arriving during the rebuild. ClickHouse’s backfilling guidance describes techniques for that operation.

One approach is a new target and a durable ingestion checkpoint. Define disjoint input ranges around the checkpoint, backfill the earlier range, and route or catch up the later range before switching readers. Use a delivery sequence or source offset that remains meaningful for late events; an event-time cutoff alone does not separate historical ingestion from live arrivals.

Overlapping input may leave the distinct purchase state unchanged while double-counting the delivery state. That is why validation must compare every measure. A rebuild that adds revenue sums would have an additional duplicate-risk surface. Keep the old target available until the new target agrees over known periods and the live handoff has been verified.

The checks I want to retain are simple to state: retries preserve logical identity, late input changes the correct day, derived counts match a known reference, and replay leaves the business answer unchanged. Reproducible Conversion Data extends that discipline to analytical results with more transformations and attribution rules.

A stream analytics pipeline becomes easier to trust when its answer can be reproduced from known events. Stable identity, explicit time semantics, and replay checks give every derived result a reference point.