---
url: 'https://altinity.com/blog/pipeline-optimization-for-clickhouse-distributed-tables-with-synchronous-inserts'
title: Pipeline Optimization for ClickHouse® Distributed Tables with Synchronous Inserts
author:
  name: Diego Nieto
  url: 'https://altinity.com/author/diego-nieto/'
date: '2026-07-27T10:15:47-07:00'
modified: '2026-07-28T03:15:23-07:00'
type: post
summary: 'Using materialized views to process data from Kafka into ClickHouse® is a common architecture. See how changing where a materialized view wrote its rows reduced ingestion lag by more than 50%.'
categories:
  - Blog
tags:
  - ClickHouse
  - Kafka
  - Materialized Views
  - Performance
image: 'https://altinity.com/wp-content/uploads/2026/07/Pipeline-Optimization.png'
published: true
---

# Pipeline Optimization for ClickHouse® Distributed Tables with Synchronous Inserts

**TL;DR:** A Kafka to ClickHouse pipeline had ingestion lag stuck at about 8.6 seconds, even with a healthy cluster and idle CPU. The cause was synchronous inserts into `Distributed` tables. Each materialized view write had to wait for a remote shard to acknowledge it, adding a fixed network round-trip of about 3.3 seconds per execution. The `system.query_views_log` table made this visible. Redirecting one hot materialized view to write to its `_local` table instead of the `Distributed` wrapper cut end-to-end lag from 8.13 seconds to 3.87 seconds on average, roughly halving p95 and p99, while keeping the stability of synchronous inserts.

A streaming pipeline was stable but it had ingestion lag. The solution turned out to be a single change to where one materialized view wrote its rows, and the metrics showed exactly where to make it.

## Introduction

A [Kafka to ClickHouse pipeline](https://altinity.com/blog/2020-5-21-clickhouse-kafka-engine-tutorial) had ingestion lag stuck at about 8.6 seconds, the cluster was healthy and CPU was idle. Tuning batch sizes did not help and consolidating low-volume topics did not help either. After some more digging and checking, the problem was that materialized views were inserting into `Distributed` tables while the cluster was using synchronous inserts. Each insert had to wait for a remote shard to acknowledge it before the view could finish.

Table `system.query_views_log` made that visible. The heavy views were spending about 3.3 seconds per execution, almost all of it in network wait, with CPU at only a few milliseconds. That is a fixed round-trip cost and it does not go away when you reduce volume. So the solution was simple: One hot materialized view was changed to write to its `_local` table instead of the `Distributed` wrapper.

After that change, end-to-end lag dropped from 8.13 seconds to 3.87 seconds on average. `p95` and `p99` were roughly cut in half as well. The result held for more than 24 hours at higher traffic, while keeping the operational stability of synchronous inserts. If you run a sharded ClickHouse cluster and have streaming lag that will not move no matter how much you tune throughput, this pattern is worth checking first.

![](https://altinity.com/wp-content/uploads/2026/07/01-Latency-KPIs.png)

## The Setup

The pipeline is a straightforward Kafka to ClickHouse flow. Quotes land in Kafka, a materialized view consumes them into a `Distributed` table, `quotes.ods_quotes`, and a second materialized view enriches them into another `Distributed` table, `quotes.quotes_eod`. The cluster has two shards and two replicas per shard. The heavy write targets are all `Distributed` tables over `ReplicatedMergeTree` local tables and this detail matters because a `Distributed` table does not store data itself, it routes inserted rows to the shard that owns them.

[Distributed tables](https://kb.altinity.com/altinity-kb-setup-and-maintenance/altinity-kb-data-migration/distributed-table-cluster/) can work in 2 modes. Default is asynchronous mode: an insert writes temporary files into a local spool directory and returns ACK to client right away, without blocking any operation. A background sender later reads those files and ships them to the destination shards. You can see the pending work in `system.distribution_queue`. In synchronous mode, the insert pushes rows to the destination shards and waits until they acknowledge receipt, like a normal `INSERT`. No spool files are created or background queue is involved. By the time the insert returns, the data is already on the shards.

This cluster ran in synchronous mode (`distributed_foreground_insert = 1`) on purpose. It hosts many `Distributed` tables across many tenant pipelines, all fed by high-frequency streaming inserts. Under asynchronous mode, those spooled files pile up faster than the background sender can drain them. That background sender uses a shared thread pool controlled by `background_distributed_schedule_pool_size`, which defaults to `16` and those threads are shared across all `Distributed` tables on the server, not reserved per table. With enough tables and enough insert frequency, the sender falls behind; even after increasing the pool to `32`, the queue still ran away to roughly 525K files and tens of GiB across the nodes before the change was rolled back. Synchronous inserts avoided that spool problem entirely generating stability at the cost of a fixed round-trip per insert. That is the tradeoff.

To measure the impact, the pipeline used two timestamps on each row:

- `kafka_received_at`: when the message arrived in Kafka

- `ch_received_at`: a `DEFAULT now()` value recorded when ClickHouse consumed the row.

Lag was measured as `ch_received_at` – `kafka_received_at`. That number stayed around 8.6 seconds. The real question was not whether the pipeline was slow. It was where those 8.6 seconds were actually going.

## Letting The Metrics Answer

Usually the first suspects are row volume, CPU, merges, or disk pressure but those metrics pointed somewhere else. [ClickHouse records materialized view](https://altinity.com/blog/performance-impact-of-materialized-views-in-clickhouse) executions in `system.query_views_log`, including `ProfileEvents`. That makes it possible to break view time into wall-clock duration, network wait, CPU time, and other components. This query was enough to identify the problem:

```
SELECT
  view_name,
  count() AS execs,
  round(avg(view_duration_ms), 1)                                       AS wall_ms,
  round(avg(ProfileEvents['NetworkReceiveElapsedMicroseconds']/1000),1) AS net_ms,
  round(avg(ProfileEvents['OSCPUVirtualTimeMicroseconds']/1000), 2)     AS cpu_ms
FROM clusterAllReplicas('{cluster}', system.query_views_log)
WHERE event_time >= now() - INTERVAL 48 HOUR
  AND view_name IN ('quotes.ods_quotes_mv', 'quotes.quotes_eod_mv')
  AND status = 'QueryFinish'
GROUP BY view_name;
```

Here’s where the time actually went, as per materialized-view execution, before the change, from `system.query_views_log`. The views weren’t busy — they were waiting on remote shard acknowledgements. CPU time is a rounding error. 

![](https://altinity.com/wp-content/uploads/2026/07/02-Where-The-Time-Went-2.png)

The result was clear. Each heavy view spent about 3.3 seconds per execution, and almost all of that time was network wait. CPU time was only a few milliseconds. `net_ms` being higher than `wall_ms` is expected here. A synchronous insert into a `Distributed` table fans out to multiple destination shards in parallel, and the network wait counters accumulate per-shard waits. The totals can add up to more than wall-clock time even though the waits overlap.

What matters is the shape of the result: the views were not busy doing work. They were mostly waiting for remote shard acknowledgements. We also checked the same views across a roughly 60x throughput range, from 13 to 778 rows per second. The per-execution cost stayed flat. That is a strong sign of a fixed per-insert round-trip, not a row-processing problem.

## The Solution

Once the cost is identified as the cross-shard round-trip, the obvious next move is to get that round-trip off the critical path. The enrichment view was changed to write to its local table, `quotes.quotes_eod_local`, instead of the `Distributed` wrapper. That means the write stays on the consuming node and does not wait for a cross-shard acknowledgement. [Replication](https://altinity.com/blog/setting-up-cross-region-clickhouse-replication-in-kubernetes) still handles consistency and durability on the local table.

![](https://altinity.com/wp-content/uploads/2026/07/05-Write-path-Fan-out-1024x652.png)

In the figure, this is the difference between the red fan-out tree branching twice and branching only once. Before the change, `ods_quotes` fans out to the shards, and then each shard-level materialized view writes into `quotes_eod` as another `Distributed` table, which triggers a second synchronous fan-out. After the change, the first fan-out still happens, but each shard-level materialized view writes into `quotes_eod_local` on the same shard, so the second red tree disappears and only background replication remains.

We could apply this approach because the target `Distributed` table used `cityHash64(...)` on a high-cardinality id, so shard assignment was already pseudo-random and primarily distribution-oriented. That is an important boundary condition. If a downstream table is sharded by tenant, account, or another locality-sensitive key, switching an MV to write directly to `_local` changes the placement model and may require a broader redesign rather than a simple latency optimization.

Also this cluster was only 2 shards. Imagine the fan out and lago for 5 or 10 shard cluster.

## The Result

The changed view stopped paying the network cost, and the pipeline KPI moved immediately. This was the end-to-end lag on `quotes.ods_quotes`, measured as `ch_received_at - kafka_received_at`, comparing 24 hours before and after the change:

```
SELECT
  multiIf(ch_received_at < '2026-06-03 09:30:00', 'before', 'after')  AS window,
  count()                                                             AS rows,
  round(avg(ch_received_at - kafka_received_at), 2)                   AS avg_s,
  round(quantile(0.95)(ch_received_at - kafka_received_at), 2)        AS p95,
  round(quantile(0.99)(ch_received_at - kafka_received_at), 2)        AS p99
FROM quotes.ods_quotes
WHERE ch_received_at BETWEEN '2026-06-02 09:30:00' AND '2026-06-04 07:30:00'
GROUP BY window ORDER BY window;
```

![](https://altinity.com/wp-content/uploads/2026/07/03-End-to-end-Lag-4.png)

And now the mechanism, view by view, as per-execution wall time. The changed view’s cross-shard round-trip vanishes, and the untouched first hop becomes the remaining bottleneck:

![](https://altinity.com/wp-content/uploads/2026/07/04-The-Mechanism-3.png)

One thing also became clearer after the change: the first hop, which still writes to a `Distributed` table, became the remaining bottleneck. Its per-execution wall time went up because the consumer was no longer blocked downstream and started cycling more often on smaller batches. The fixed round-trip cost did not change. It simply became more visible.

## Apply This to Your Own Pipeline

If you want to check for the same issue in your own pipeline, this is the order that worked here.

### 1. Measure lag directly

Use a source timestamp and an ingest timestamp if you have them. If the lag stays roughly flat when traffic goes up and down, that points to a fixed per-insert cost rather than a throughput bottleneck.

### 2. Break down materialized view time with `system.query_views_log`

Run the per-view query and compare `wall_ms`, `net_ms`, and `cpu_ms`.

The pattern to look for is:

- `net_ms` close to or higher than `wall_ms`

- `cpu_ms` close to zero

- no meaningful disk or Keeper time if you include those counters

That means the view is mostly waiting on the network.

### 3. Check whether the hot views write into `Distributed` tables

Use these queries:

```
SELECT database, name
FROM system.tables
WHERE engine = 'Distributed';

SELECT name, value
FROM system.settings
WHERE name IN ('distributed_foreground_insert', 'insert_distributed_sync');
```

If the slow view writes to a `Distributed` target and the cluster is using synchronous inserts, you likely found the bottleneck.

### 4. Compare with a local-target view on the same cluster

Find a materialized view that writes to a `MergeTree` or `ReplicatedMergeTree` target and compare its timings. If it runs in single-digit milliseconds with no network wait, that is a strong control case. It shows that the cluster itself is not the problem.

### 5. Check sharding and dedup before redirecting writes

This is the main safety check. Writing to `_local` changes where rows land. Instead of being routed by the `Distributed` table sharding key, rows stay on the node that consumed them.

That is usually fine only if:

- the pipeline does not require cross-shard routing at write time

- rows that must deduplicate together still land on the same shard.

For `ReplacingMergeTree` targets, make sure the producer partition key and the original `Distributed` sharding key line up closely enough for that guarantee to hold.

### 6. Repoint the hot materialized view to `_local`

Recreate the materialized view so it writes to the local table instead of the `Distributed` wrapper. This is a metadata change in the write target. It does not require a server restart.

### 7. Re-measure over equal time windows

Run the lag KPI again. Run the `query_views_log` query again. Split the results at the change time. The expected result is that network wait drops toward zero for the changed view and the end-to-end lag drops with it.

If another stage in the chain still writes to `Distributed`, repeat the same process there.

## Closing Notes

Synchronous `Distributed` inserts can be the right operational choice, but on a hot streaming path they can also become the main source of lag. In this case, `system.query_views_log` showed that one materialized view was mostly waiting for remote shard acknowledgements, and changing its target from `Distributed` to `_local` removed about half of the end-to-end delay without giving up the stability of synchronous inserts.