DEV Community: Maksim Grishchenko

Generating Kafka ClickHouse pipelines from a JSON sample

Maksim Grishchenko — Thu, 22 Oct 2026 20:12:32 +0000

This post isn't about ClickHouse internals. It's about a part of my day-to-day work that kept eating my time. You want to build something great, and instead you sit there struggling to understand what is going on. We consume a lot of data straight from Kafka into ClickHouse, using ClickHouse's own Kafka consumer (the Kafka table engine) and materialized views. Data is constantly flowing, and the message format keeps changing. Every new topic meant the same job:

find a message on the topic and check its fields,
write several migrations that have to be applied in a particular order (we use golang-migrate),
apply them and check that data shows up (not on prod, we have a separate environment for that).

For a flat message with a few fields that's fine. When the message has nested objects or arrays, it gets slow and easy to get wrong. At some point I noticed how much time I was spending on it and started looking for a faster way.

TL;DR

Writing the Kafka → ClickHouse migrations by hand was the slow part, especially for nested messages. A shell script around clickhouse local helped, but it needed a 663 MB binary and silently mixed up nested fields. clickforge generates the same five statements in milliseconds with no dependencies. It doesn't make changing a running flow any easier, which is still the hard part.

The flow: streams, raw, datalake

The Kafka table is a consumer, not storage. A plain SELECT from it reads messages and moves the consumer group on, so each message can be read only once (docs). That's why nobody queries streams directly. Once a materialized view is attached, ClickHouse reads the topic in the background and pushes each batch of messages through the view into the target table. We want to keep every message as it arrived, so the first view writes into raw, and a second view on raw fills the typed datalake table.

Each topic gets three tables and two materialized views:

Kafka topic
    ↓
streams.video_events     Kafka engine table: reads the topic, stores nothing
    ↓  streams.video_events_mv
raw.video_events         the original message as a string, plus Kafka metadata
    ↓  raw.video_events_mv
datalake.video_events    typed columns, extracted from the message with JSONExtract

The migration, layer by layer

One topic is one up migration with five statements, run in this order. The SQL below is what clickforge generates for video_events.json, a sample of video player events (the shell script produced the same five statements). I've cut the column lists down where they repeat.

The first statement creates the consumer. It has a single message column, and kafka_format = 'RawBLOB' puts the whole Kafka message into it as one string, so nothing is parsed at this stage. Kafka(kafka) takes the broker settings from a named collection called kafka. I've replaced our internal topic and consumer group names with plain ones.

CREATE TABLE IF NOT EXISTS `streams`.`video_events` ON CLUSTER clickhouse_datalake
(
    `message` String
)
    ENGINE = Kafka(kafka) SETTINGS kafka_topic_list =
    'video_events', kafka_group_name =
    'clickhouse-video_events', kafka_format = 'RawBLOB';

Credentials don't go into the migration. They live in the named collection, which is created once per cluster, either in the server config or with SQL:

CREATE NAMED COLLECTION kafka AS
    kafka_broker_list = 'kafka1:9092,kafka2:9092',
    kafka_security_protocol = 'sasl_ssl',
    kafka_sasl_mechanism = 'SCRAM-SHA-512',
    kafka_sasl_username = 'clickhouse',
    kafka_sasl_password = '...';

This way the migrations can sit in git without passwords, and every topic table just says Kafka(kafka).

The second is raw, a normal replicated table that stores the message unchanged, together with where it came from. _key, _offset, _partition, _timestamp_ms and _topic are virtual columns of the Kafka engine, and _row_created is the time ClickHouse wrote the row. If the schema changes or the typed table turns out to be wrong, I can rebuild datalake from raw without reading the topic again.

CREATE TABLE IF NOT EXISTS `raw`.`video_events` ON CLUSTER clickhouse_datalake
(
    `message`       String,
    `_key`          String,
    `_offset`       UInt64,
    `_partition`    UInt64,
    `_timestamp_ms` DateTime64(3),
    `_topic`        LowCardinality(String),
    `_row_created`  DateTime DEFAULT nowInBlock()
)
    ENGINE = ReplicatedMergeTree('/clickhouse/{cluster}/tables/raw/video_events/{shard}', '{replica}')
    PARTITION BY toYYYYMM(_row_created)
    ORDER BY _row_created
    SETTINGS index_granularity = 8192;

The third is datalake, the table people query. Each field of the message becomes a typed column, and the Kafka timestamp, topic and write time are carried over from raw. It's partitioned by the month of the Kafka timestamp, so a message lands in the month of its Kafka timestamp, not the month it was loaded.

CREATE TABLE IF NOT EXISTS `datalake`.`video_events` ON CLUSTER clickhouse_datalake
(
    `event_id` String,
    `user_id` Int64,
    ...
    `client_ip` Nullable(String),
    `country_code` Nullable(String),
    `platform` String,
    `_timestamp_ms` DateTime64(3),
    `_topic`        LowCardinality(String),
    `_row_created`  DateTime DEFAULT nowInBlock()
)
    ENGINE = ReplicatedMergeTree('/clickhouse/{cluster}/tables/datalake/video_events/{shard}', '{replica}')
    PARTITION BY toYYYYMM(_timestamp_ms)
    ORDER BY _timestamp_ms
    SETTINGS index_granularity = 8192;

The fourth connects raw to datalake. A materialized view works like an insert trigger: every block of rows inserted into raw goes through this SELECT, and the result is inserted into datalake. This is where the message gets parsed, one JSONExtract per column.

CREATE MATERIALIZED VIEW IF NOT EXISTS `raw`.`video_events_mv`
    ON CLUSTER clickhouse_datalake TO `datalake`.`video_events` AS
SELECT * FROM (
    SELECT
        JSONExtract(message, 'event_id', 'String') AS `event_id`,
        JSONExtract(message, 'user_id', 'Int64') AS `user_id`,
        ...
        JSONExtract(message, 'client_ip', 'Nullable(String)') AS `client_ip`,
        `_timestamp_ms`,
        `_topic`,
        `_row_created`
    FROM `raw`.`video_events`
) SETTINGS cast_keep_nullable = 1;

cast_keep_nullable = 1 keeps Nullable when a value is cast to another type. By default it's dropped: on 26.9, CAST(toNullable(toInt32(0)) AS String) returns String, and with the setting on it returns Nullable(String).

The last statement connects the consumer to raw, and the moment it exists, ClickHouse starts reading the topic. _timestamp_ms is Nullable on the Kafka table, so assumeNotNull turns it into the plain DateTime64(3) that raw expects.

CREATE MATERIALIZED VIEW IF NOT EXISTS `streams`.`video_events_mv`
    ON CLUSTER clickhouse_datalake
    TO `raw`.`video_events` (
        `message`        String,
        ...
        `_row_created`   DateTime
    ) AS
SELECT
    message,
    _key,
    _offset,
    _partition,
    assumeNotNull(_timestamp_ms) AS _timestamp_ms,
    _topic,
    nowInBlock() AS _row_created
FROM `streams`.`video_events`;

That's why the order matters. A materialized view only sees rows inserted after it was created, and it doesn't go back for the ones already in the table.

Changing a running flow is harder than creating it. To change the schema, I drop the streams table and its view first so the consumer stops, change the tables, and create everything again in the right order. Get the order wrong, or forget to update a view, and the tables just stay empty. There's no error to tell you why, and you can't simply look at the topic from ClickHouse, because reading the Kafka table consumes the messages. clickforge doesn't solve this. It only removes the bottleneck of writing the DDL.

First attempt: a shell script

My first fix was a shell script. It takes a JSON sample of the topic's messages and lets ClickHouse work out the schema:

./create_ddl_for_kafka.sh video_events.json

Inside, it runs clickhouse local to create a temporary table from the file with schema inference, prints SHOW CREATE TABLE, and then rewrites that output with sed and awk into the five statements. The column list goes into the datalake table, and each column becomes a JSONExtract(message, 'field', 'Type') line in the materialized view. It took a heavy session with an LLM to make it work, but I was serious, I wanted my time back.

It worked, but it was never convenient. It needs a ClickHouse binary next to it (./clickhouse.0, 663 MB on my laptop), and every column comes out Nullable, because the script turns on schema_inference_make_columns_nullable. A run took about six seconds. The engine is fixed to ReplicatedMergeTree, so there's no way to pick another one, and the awk step that rewrites types gets the brackets wrong once a type is nested.

At work, a message can be a hundred lines of loosely structured, nested JSON, and it has to land in a flat, typed table. The bigger problem showed up when I ran it on a message with nested fields for this post:

{"order_id": 1, "customer": {"id": 7, "address": {"city": "Berlin"}}, "items": [{"sku": "A1", "qty": 2}], "tags": {"source": "app", "campaign": "fall"}}

clickhouse local infers named tuples for the objects, for example tags Tuple(campaign Nullable(String), source Nullable(String)). The awk step strips everything in backticks from the type, so the names inside the tuple are lost and the view ends up with this:

JSONExtract(message, 'tags', 'Tuple( Nullable(String), Nullable(String))') AS tags

Without names, JSONExtract fills the tuple by position. ClickHouse sorted the inferred fields alphabetically (campaign, source), but the message has them the other way round. Here is what the generated extracts return for the message above on 26.9:

customer: ((NULL),NULL)
items:    [(NULL,'2')]
tags:     ('app','fall')

All of customer is lost, qty landed in the sku position, and tags has campaign = 'app' and source = 'fall'. None of these extracts raises an error, so nothing tells you the columns are wrong.

clickforge

So I wrote a CLI, clickforge. Kind of a weird name, but I like it. It's a single binary with no ClickHouse dependency, and it generates the same flow from the same JSON sample:

clickforge kafka video_events.json

13 columns inferred from video_events

  event_id              String
  user_id               Int64
  ...
  client_ip             String
  country_code          String
  platform              String

Written: ./video_events_up.sql
Written: ./video_events_down.sql

That took 7 ms. video_events_up.sql holds the five statements in the order above, and video_events_down.sql drops them again. A field is Nullable only if it's missing or null in some message, so in this sample only client_ip and country_code are Nullable, because the last event has neither.

For the nested message, clickforge keeps nested objects as JSON text and turns flat objects into a Map:

customer: {"id":7,"address":{"city":"Berlin"}}
items:    [{'sku':'A1','qty':'2'}]
tags:     {'source':'app','campaign':'fall'}

Nothing is lost or swapped, but it isn't fully typed either. qty is the string '2', and customer needs JSONExtract at query time.

Closing words

clickforge is a small CLI, and it doesn't solve everything. Changing a running flow is still manual and still easy to get wrong. What it gives me is the thing I was missing: a fast, correct scaffold for a new topic, so I can apply the migrations and see data in the tables a few seconds later. Not a bad day if it saves me a couple of hours of work. I'm still working on the CLI and trying to make it better.

If you want to try it yourself, the JSON samples and the JSONExtract checks are next to this post: video_events.json, nested.json, extract-check.sql for the nested test, and flow-check.sql for the raw → datalake layers.

When ClickHouse merges change your query results

Maksim Grishchenko — Tue, 20 Oct 2026 07:16:58 +0000

In Part 1 I followed an order from INSERT through a merge. That answered two questions that came up during coffee breaks at work. This post tackles the third:

"Why doesn't my SELECT show the values I expect?"

I usually hear this question from people using SummingMergeTree, AggregatingMergeTree, or CollapsingMergeTree. In these engines, a merge can change the rows themselves: it sums them, collapses them, or removes them. As a result, the same query can return different results depending on how many merges have occurred.

If you skipped Part 1, three things from it matter here. Each inserted block typically creates one or more new parts, and ClickHouse never modifies a part in place; a merge writes a new, larger part and removes the old ones later. Merges run in the background whenever the merge selector decides they're worthwhile, so you can't predict when two rows will end up in the same part. And merges never cross partition boundaries.

Everything here ran on ClickHouse 26.9.

1. When merges change the answer

In Part 1, ReplacingMergeTree kept the latest version of each order. It's one of several engines that apply special handling when rows with the same sorting key meet. ORDER BY determines which rows belong together; the engine determines what happens to them.

Engine	Rows with the same `ORDER BY` key become…
`ReplacingMergeTree(ver)`	the row with the highest `ver` (or the last inserted)
`SummingMergeTree`	one row with numeric columns summed
`AggregatingMergeTree`	one row with aggregate states merged
`CollapsingMergeTree(sign)`	nothing, if a `+1` row is cancelled by a `-1` row
`VersionedCollapsingMergeTree(sign, ver)`	the same, matched by version, in any order
`CoalescingMergeTree` (25.6+)	one row with the latest non-NULL value of each column

The important question is when that transformation happens. The engine's merge logic is applied at three points, using the same algorithm classes:

On insert, within a single insert block. This is why newly inserted parts for these engines start at level 1. MergeTreeDataWriter::mergeBlock selects the algorithm (source):

   switch (merging_params.mode)
   {
       /// There is nothing to merge in single block in ordinary MergeTree
       case MergeTreeData::MergingParams::Ordinary:
           return nullptr;
       case MergeTreeData::MergingParams::Replacing:
           return std::make_shared<ReplacingSortedAlgorithm>(/* ... */);
       case MergeTreeData::MergingParams::Collapsing:
           return std::make_shared<CollapsingSortedAlgorithm>(/* ... */);
       case MergeTreeData::MergingParams::Summing:
           return std::make_shared<SummingSortedAlgorithm>(/* ... */);
       // ... Aggregating, VersionedCollapsing, Graphite, Coalescing
   }

During a background merge, for the selected parts within a single partition (MergeTask.cpp).
At query time, when FINAL is used, for the parts read by the query (MergeTreeFinalMerge.cpp).

Between these stages, rows with the same key can remain in separate parts, unprocessed. A plain SELECT sees whatever state the background merges have reached so far. Here's what that means for each engine. In the SummingMergeTree and CollapsingMergeTree examples, each INSERT starts its own part, and I then force a full merge with OPTIMIZE TABLE ... FINAL. The AggregatingMergeTree example uses batches instead.

SummingMergeTree: zero rows disappear

SummingMergeTree is a table of counters. The official documentation describes it as follows:

The difference is that when merging data parts for SummingMergeTree tables ClickHouse replaces all the rows with the same primary key (or more accurately, with the same sorting key) with one row which contains summed values for the columns with the numeric data type.

Instead of keeping one row per event, the table combines values for each key as parts merge. A typical use case is daily statistics, such as page views per page per day, often populated by a materialized view. The documentation also recommends retaining the raw events in a plain MergeTree table alongside it, so a mistake in the sorting key doesn't result in data loss.

Here's the table I played with:

CREATE TABLE page_stats (day Date, page String, views Int64, title String)
ENGINE = SummingMergeTree ORDER BY (day, page);

ORDER BY (day, page) defines the sorting key: one counter per page per day.
views is numeric, so merges sum it.
title is a string, so it can't be summed. It's included to show what happens to non-numeric columns.

I stopped merges and ran four INSERT statements. The last one is a correction that cancels out the views for /promo:

SYSTEM STOP MERGES page_stats;

INSERT INTO page_stats VALUES ('2026-10-10', '/home',  10, 'Home');
INSERT INTO page_stats VALUES ('2026-10-10', '/home',   5, 'Home (new title)');
INSERT INTO page_stats VALUES ('2026-10-10', '/promo',  7, 'Promo');
INSERT INTO page_stats VALUES ('2026-10-10', '/promo', -7, 'Promo');   -- a correction

Before the merge, each INSERT is its own part, and nothing has been summed yet:

SELECT _part, page, views, title FROM page_stats ORDER BY _part;

┌─_part─────┬─page───┬─views─┬─title────────────┐
│ all_1_1_1 │ /home  │    10 │ Home             │
│ all_2_2_1 │ /home  │     5 │ Home (new title) │
│ all_3_3_1 │ /promo │     7 │ Promo            │
│ all_4_4_1 │ /promo │    -7 │ Promo            │
└───────────┴────────┴───────┴──────────────────┘

A plain SELECT doesn't sum anything, either. It returns all four rows, with two rows per page:

SELECT * FROM page_stats ORDER BY page, views;

┌────────day─┬─page───┬─views─┬─title────────────┐
│ 2026-10-10 │ /home  │     5 │ Home (new title) │
│ 2026-10-10 │ /home  │    10 │ Home             │
│ 2026-10-10 │ /promo │    -7 │ Promo            │
│ 2026-10-10 │ /promo │     7 │ Promo            │
└────────────┴────────┴───────┴──────────────────┘

To get the totals, the query has to sum them itself:

SELECT page, sum(views) AS total FROM page_stats GROUP BY page ORDER BY page;

┌─page───┬─total─┐
│ /home  │    15 │
│ /promo │     0 │
└────────┴───────┘

Next, I switched merges back on and forced a merge:

SYSTEM START MERGES page_stats;
OPTIMIZE TABLE page_stats FINAL;

SELECT _part, page, views, title FROM page_stats ORDER BY _part;

┌─_part─────┬─page──┬─views─┬─title─┐
│ all_1_4_2 │ /home │    15 │ Home  │
└───────────┴───────┴───────┴───────┘

The four parts became one part, all_1_4_2: blocks 1 through 4, merged at level 2. The two /home rows became a single row with 15 views, while the /promo rows disappeared.

Both behaviors are documented in the algorithm's header (source):

/** Merges several sorted inputs into one.
  * For each group of consecutive identical values of the primary key (the columns by which the data is sorted),
  *  collapses them into one row, summing all the numeric columns except the primary key.
  * If in all numeric columns, except for the primary key, the result is zero, it deletes the row.
  */

The first thing that bit me is the obvious one: a key whose numeric values all sum to 0 can disappear. The /promo views summed to 0, so the merge deleted the row. For page-view counts that's fine, but if the table stores account balances, an account with a zero balance quietly vanishes from it.

The second one I didn't expect at all. I assumed that a query which sums the rows itself is safe no matter what the merges do, and it mostly is, but not about which rows come back. Before the merge, the GROUP BY query above returned /promo with a total of 0; after the merge, the same query returns only /home. To keep the output the same either way, add HAVING sum(views) != 0, which hides zero totals before a merge too. (My first version of that query failed with ILLEGAL_AGGREGATION, because I'd named the result views, the same as the column, and ClickHouse then read sum(views) in HAVING as a sum of the sum. Hence total.)

The third is the title column, which kept 'Home' and threw away 'Home (new title)'.
Non-numeric columns outside the sorting key aren't summed, and the merge keeps the value from the first row, which is the row from the part with the lowest block number. Insert the two rows in reverse order and 'Home (new title)' wins instead. (Thanks to a reader in the comments for suggesting that test.) The docs call this value arbitrary, and with concurrent inserts "first" means first to commit, not first sent, so I wouldn't rely on it. If you care which title you get, store an updated_at timestamp with each row and query argMax(title, updated_at): the newest title wins, however the parts were merged.

AggregatingMergeTree: a plain column changes its value after a merge

CREATE TABLE daily
(
    day       Date,
    cnt_plain UInt64,                                -- the mistake
    cnt       SimpleAggregateFunction(sum, UInt64),  -- the fix
    users     AggregateFunction(uniq, UInt64)
)
ENGINE = AggregatingMergeTree ORDER BY day;

This is where 26.9 surprised me: the CREATE TABLE statement fails before I can even insert anything.

Column(s) cnt_plain of the AggregatingMergeTree table are neither part of the sorting key nor
aggregate measures (AggregateFunction or SimpleAggregateFunction). During background merges, rows
with the same value of the sorting key are collapsed into one, so these columns keep an arbitrary
value, which silently produces wrong results ...

The check was added in 26.7, and older versions accepted this schema without a word. I only found out because my experiment script, written the old way, failed on 26.9. To show what the check prevents, I recreated the table with SETTINGS allow_dimensions_outside_sorting_key = 1 (the old behavior) and inserted two batches containing 10 and 5 events:

Query	Before merge	After merge
`sum(cnt_plain)`	15	10
`sum(cnt)`	15	15
`uniqMerge(users)`	12	12

So the same query gives 15 before the merge and 10 after it. cnt_plain retains the first row's value, as the algorithm's header explains (source): "For columns that are not part of the primary key and which do not have the AggregateFunction type, when merged, the first value is selected." SimpleAggregateFunction(sum, ...) and AggregateFunction(...) columns combine correctly across merge states.

The same principle applies when querying: always re-aggregate with functions such as sum(cnt) and uniqMerge(users), grouped by the sorting key. Some rows may already have been merged while others haven't; re-aggregation makes the result independent of that state.

CollapsingMergeTree: the order of parts matters

CollapsingMergeTree models state changes as a log. To update a row, insert a copy of the old state with sign = -1 to cancel it, followed by the new state with sign = 1:

CREATE TABLE user_state (user_id UInt64, page_views UInt32, sign Int8)
ENGINE = CollapsingMergeTree(sign) ORDER BY user_id;

-- user 1: state, then cancel + new state (the documented pattern)
INSERT INTO user_state VALUES (1, 5, 1);
INSERT INTO user_state VALUES (1, 5, -1), (1, 6, 1);

-- user 2: the cancel row arrives BEFORE the state it cancels
INSERT INTO user_state VALUES (2, 5, -1);
INSERT INTO user_state VALUES (2, 5, 1);

OPTIMIZE TABLE user_state FINAL;

┌─_part─────┬─user_id─┬─page_views─┬─sign─┐
│ all_1_4_2 │       1 │          6 │    1 │
│ all_1_4_2 │       2 │          5 │   -1 │
│ all_1_4_2 │       2 │          5 │    1 │
└───────────┴─────────┴────────────┴──────┘

User 1 collapsed as expected. User 2's pair survived a full merge. The reason is explained in the algorithm's rules (source):

/** ...
  * If the number of positive and negative rows is the same, and the last row is positive, then the first negative and last positive rows are written.
  * If the number of positive and negative rows is the same, and the last line is negative, it writes nothing.
  * ...
  */

Within each sorting key, rows are processed in part order (by block number). In this example, user 2's -1 part came first, so the group ended with a positive row and both rows survived. With parallel writers, retries, or Kafka partitions, cancellation rows can arrive before the states they cancel.

You can keep the query correct in two ways. The first is to let sign do the work in the query, which gives the right answer whether or not the rows have collapsed yet:

SELECT user_id, sum(toInt64(page_views) * sign) AS page_views
FROM user_state
GROUP BY user_id
HAVING sum(sign) > 0;

This returns only user 1, with 6 page views, because user 2's rows cancel each other out in the sums.

The second is to switch to VersionedCollapsingMergeTree(sign, version), which I'd honestly prefer whenever the writers aren't strictly ordered. It matches a -1 row with a +1 row of the same version, regardless of their order (source). Repeating the out-of-order inserts for user 2 with version = 1 on both rows left no rows after OPTIMIZE TABLE ... FINAL.

The rule for all of them

Write every query so it returns the same result whether no merges, some merges, or all possible merges have occurred. In practice:

Engine	Query pattern that's always correct
`ReplacingMergeTree`	`FINAL`, or `argMax(col, ver) ... GROUP BY key`
`SummingMergeTree`	`sum(a), sum(b) ... GROUP BY key HAVING sum(a) != 0 OR sum(b) != 0` (a row is deleted only when all its numeric columns sum to 0)
`AggregatingMergeTree`	`-Merge` combinators and `sum`/`max`/… for `SimpleAggregateFunction`, `GROUP BY key`
`CollapsingMergeTree`	`sum(col * sign) ... GROUP BY key HAVING sum(sign) > 0`
`VersionedCollapsingMergeTree`	Same as Collapsing

Remember the limits: merges never cross partition boundaries, and they run only when the merge selector decides they're worthwhile. Treat merges as a storage optimization, and make your query results independent of merge state.

2. Deleting and updating in place

Parts are immutable, so how do DELETE and UPDATE work? To demonstrate the effect on a realistically sized part, I used a MergeTree table containing one million orders in a single part, all_1_1_1.

Lightweight DELETE:

DELETE FROM orders_mt WHERE order_id < 10;

┌─name────────┬────rows─┬─size─────┐
│ all_1_1_1_2 │ 1000000 │ 3.94 MiB │
└─────────────┴─────────┴──────────┘

count() now returns 999,990, but the part still physically contains 1,000,000 rows. The part name gained a fifth field, _2, which is the mutation version. system.mutations shows the operation that ran: UPDATE _row_exists = 0 WHERE order_id < 10. A lightweight DELETE marks rows in the hidden _row_exists column but leaves their data in place. The rows are physically removed during later merges.

ALTER TABLE ... DELETE (a classic mutation):

ALTER TABLE orders_mt DELETE WHERE order_id < 20 SETTINGS mutations_sync = 1;

┌─name────────┬───rows─┬─size─────┐
│ all_1_1_1_3 │ 999980 │ 3.94 MiB │
└─────────────┴────────┴──────────┘

The part was rewritten, and 999,980 rows are now physically present. On a real table, this rewrites every affected part and can generate substantial I/O.

Lightweight UPDATE (requires enable_block_number_column = 1 and enable_block_offset_column = 1 on the table):

UPDATE orders_mt SET status = 'paid' WHERE order_id = 500;

┌─name───────────────────────────────────────────────┬───rows─┬─size─────┐
│ all_1_1_1_3                                        │ 999980 │ 3.94 MiB │
│ patch-ca8f64b62d9505701c71c81624162ef0-all_5_5_0_4 │      1 │ 1.05 KiB │
└────────────────────────────────────────────────────┴────────┴──────────┘

Instead of rewriting a 4 MiB part to change one value, ClickHouse wrote a 1 KiB patch part containing one row. Reads apply the patch on the fly, and later merges fold it into the main part.

Operation	What it writes	Data physically removed or changed
`DELETE FROM`	A row mask (`_row_exists`)	On later merges
`ALTER ... DELETE` / `UPDATE`	Full rewrite of affected parts	When the mutation completes (asynchronously by default)
`UPDATE ... SET`	A small patch part	On later merges
`ALTER ... DROP PARTITION`	Removes the partition's data parts	Immediately at the table level; physical cleanup may follow

Back to the coffee break

So, "Why doesn't my SELECT show what I expect?" In SummingMergeTree, AggregatingMergeTree and CollapsingMergeTree tables, the raw rows depend on which merges have happened, so every query should give the same result whether no merges, some merges or all of them have run. In practice that means re-aggregating at query time.

A few more things I'd tell a colleague. In an AggregatingMergeTree, put every column either in ORDER BY or into an aggregate (SimpleAggregateFunction or AggregateFunction); since 26.7 ClickHouse won't let you do anything else, which I think is a good call. In a SummingMergeTree, remember that a row disappears on merge when all its numeric columns sum to zero, so think twice before storing anything where zero is a meaningful value, like balances. If cancel rows can arrive before the rows they cancel, use VersionedCollapsingMergeTree. And for deletes, use lightweight DELETE/UPDATE for targeted changes, DROP PARTITION when a whole partition goes, and classic mutations only when you actually want the full rewrite.

Follow one row: what ClickHouse does with your INSERT

Maksim Grishchenko — Sun, 18 Oct 2026 06:05:34 +0000

I work as a data infrastructure engineer at a company where ClickHouse comes up at almost every coffee break. The questions I get are usually some version of these:

"Why do I still see the old value? I'm using ReplicatedReplacingMergeTree."
"Why are we getting Too many parts errors?"
"Why doesn't my SELECT show the values I expect?"

At some point I noticed that all of them come from the same gap. The people asking had never seen the whole flow, from the moment a row is inserted to the moment it's merged, deduplicated or deleted.

So one day I looked at our ClickHouse cluster and remembered the line from the Inside Out trailer:

Do you ever look at someone and wonder: what is going on inside their head?

So I decided to look inside ClickHouse's head.

I knew the semantics. I'd read the official ClickHouse docs and plenty of blog posts. But something was missing, even for me. I decided it's better to touch it once than to hear about it a hundred times.

I followed one order through its life in ClickHouse: insert, parts, merges, updates and duplicates. I captured real output at each step and looked up in the source code why it behaves that way. Where a single row wasn't enough to show something, I used a bigger batch or a separate test table, and I say so each time.

TL;DR

"Why do I still see the old value?" The new version sits in a different part that hasn't been merged with the old one yet, and merges are lazy. If the two versions are in different partitions, they will never be merged. Read with FINAL or argMax, and keep every version of a row in the same partition.
"Why are we getting Too many parts?" A non-empty INSERT normally creates at least one part per partition. Many small INSERTs, or a partition key that's too fine, create parts faster than merges can combine them. Batch on the client, or use async inserts when many clients write at once.
"Why doesn't my SELECT show the values I expect?" In Summing, Aggregating and Collapsing tables, merges change the rows themselves, so re-aggregate in every query. Part 2 covers this in detail.

The rest of the post shows where these answers come from.

Preparations

Before we look inside ClickHouse, let's do some preparations. We'll use Docker for simplicity and one small table. Every query here is in one script, so you can check any claim yourself as you read.

docker run -d --name ch --ulimit nofile=262144:262144 clickhouse/clickhouse-server:26.9
docker exec -it ch clickhouse-client

Everything happens inside this container. The SQL runs in the clickhouse-client session you just opened. Shell commands run through docker exec ch ..., and paths like /var/lib/clickhouse/... are paths inside the container, not on your machine.

Everything here ran on ClickHouse 26.9, and several defaults changed recently, so on an older version some numbers will differ. I use the non-replicated engines to keep the setup to one container, but all of it applies to ReplicatedReplacingMergeTree too. The only difference is who decides which parts to merge: in a replicated table, a leader replica writes that decision into the replication log in Keeper (source), and every replica then runs the same merge on its own copy, or, for very large parts, downloads the result (source).

The table:

CREATE TABLE orders
(
    order_id   UInt64,
    status     LowCardinality(String),
    amount     Decimal(10, 2),
    updated_at DateTime64(3)
)
ENGINE = ReplacingMergeTree(updated_at)
PARTITION BY toYYYYMM(updated_at)
ORDER BY order_id;

When it merges parts, ReplacingMergeTree(updated_at) keeps the latest version of each order_id, because ORDER BY order_id defines which rows ClickHouse considers duplicates, while updated_at defines which duplicate wins. This partition key is a mistake on purpose. We'll come back to it.

To watch parts pile up, we stop background merges for now:

SYSTEM STOP MERGES orders;

1. An INSERT creates a part

INSERT INTO orders VALUES (1, 'created', 10.00, now64());
INSERT INTO orders VALUES (2, 'created', 10.00, now64());
INSERT INTO orders VALUES (3, 'created', 10.00, now64());

SELECT name, part_type, rows, level
FROM system.parts
WHERE table = 'orders' AND active;

┌─name─────────┬─part_type─┬─rows─┬─level─┐
│ 202610_1_1_1 │ Compact   │    1 │     1 │
│ 202610_2_2_1 │ Compact   │    1 │     1 │
│ 202610_3_3_1 │ Compact   │    1 │     1 │
└──────────────┴───────────┴──────┴───────┘

Three INSERTs gave me three parts.

What a part is

The official docs describe it like this:

The data from each table in the ClickHouse MergeTree engine family is organized on disk as a collection of immutable data parts.

A data part is created whenever a set of rows is inserted into the table.

So a table is a set of parts, and every INSERT adds new ones. To see which part each row lives in, select the virtual column _part:

SELECT _part, order_id, status FROM orders;

┌─_part────────┬─order_id─┬─status──┐
│ 202610_1_1_1 │        1 │ created │
│ 202610_2_2_1 │        2 │ created │
│ 202610_3_3_1 │        3 │ created │
└──────────────┴──────────┴─────────┘

Each part is a directory on disk. system.parts shows where:

SELECT name, rows, bytes_on_disk, active, path
FROM system.parts
WHERE table = 'orders';

┌─name─────────┬─rows─┬─bytes_on_disk─┬─active─┬─path─────────────────────────────────────────────────────┐
│ 202610_1_1_1 │    1 │          1250 │      1 │ /var/lib/clickhouse/store/621/62173928-.../202610_1_1_1/ │
│ 202610_2_2_1 │    1 │          1250 │      1 │ /var/lib/clickhouse/store/621/62173928-.../202610_2_2_1/ │
│ 202610_3_3_1 │    1 │          1250 │      1 │ /var/lib/clickhouse/store/621/62173928-.../202610_3_3_1/ │
└──────────────┴──────┴───────────────┴────────┴──────────────────────────────────────────────────────────┘

active = 1 means the part is currently part of the table. Here is what's inside one of these directories:

$ docker exec ch ls -l /var/lib/clickhouse/store/621/62173928-.../202610_1_1_1/
total 52
-rw-r----- 1 clickhouse clickhouse 379 Oct 11 12:45 checksums.txt
-rw-r----- 1 clickhouse clickhouse 254 Oct 11 12:45 columns_substreams.txt
-rw-r----- 1 clickhouse clickhouse 138 Oct 11 12:45 columns.txt
-rw-r----- 1 clickhouse clickhouse   1 Oct 11 12:45 count.txt
-rw-r----- 1 clickhouse clickhouse 161 Oct 11 12:45 data.bin
-rw-r----- 1 clickhouse clickhouse  72 Oct 11 12:45 data.cmrk4
-rw-r----- 1 clickhouse clickhouse  10 Oct 11 12:45 default_compression_codec.txt
-rw-r----- 1 clickhouse clickhouse   1 Oct 11 12:45 metadata_version.txt
-rw-r----- 1 clickhouse clickhouse  16 Oct 11 12:45 minmax_updated_at.idx
-rw-r----- 1 clickhouse clickhouse   4 Oct 11 12:45 partition.dat
-rw-r----- 1 clickhouse clickhouse  50 Oct 11 12:45 primary.cidx
-rw-r----- 1 clickhouse clickhouse 410 Oct 11 12:45 serialization.json
-rw-r----- 1 clickhouse clickhouse 536 Oct 11 12:45 statistics.packed

data.bin holds the values of all four columns, compressed. primary.cidx is the primary index and data.cmrk4 the marks, which help ClickHouse find rows quickly, and the rest is metadata: row count, column types, the partition, the min and max updated_at, and checksums of every file.

How much disk does one order cost? My first answer was 1,250 bytes, the bytes_on_disk value from system.parts, and that's what went into the first draft. Then I added up the sizes from ls -l, got 2,032, and went digging: bytes_on_disk only counts the files listed in checksums.txt (source), which leaves out a few small metadata files. The column data itself, data.bin, is 161 bytes. And the filesystem doesn't care about any of these numbers, because on this container every file takes at least one 4 KiB block:

$ docker exec ch du -k /var/lib/clickhouse/store/621/62173928-.../202610_1_1_1/
56  /var/lib/clickhouse/store/621/62173928-.../202610_1_1_1/

So one order took 161 bytes of data and 56 KiB of disk, which means a tiny part is almost all overhead.

A big INSERT turns this around. Into a copy of the same table, I inserted one million orders with a single INSERT. They all fall in October, so they became one part:

$ docker exec ch ls -l /var/lib/clickhouse/store/a7a/a7a4bb8f-.../202610_1_1_1/
total 12600
-rw-r----- 1 clickhouse clickhouse 3787369 Oct 13 07:15 amount.bin
-rw-r----- 1 clickhouse clickhouse     410 Oct 13 07:15 amount.cmrk2
...
-rw-r----- 1 clickhouse clickhouse 4004915 Oct 13 07:15 order_id.bin
-rw-r----- 1 clickhouse clickhouse     415 Oct 13 07:15 order_id.cmrk2
...
-rw-r----- 1 clickhouse clickhouse    5120 Oct 13 07:15 status.bin
-rw-r----- 1 clickhouse clickhouse      59 Oct 13 07:15 status.dict.bin
...
-rw-r----- 1 clickhouse clickhouse 5019580 Oct 13 07:15 updated_at.bin
-rw-r----- 1 clickhouse clickhouse     429 Oct 13 07:15 updated_at.cmrk2

$ docker exec ch du -k /var/lib/clickhouse/store/a7a/a7a4bb8f-.../202610_1_1_1/
12604   /var/lib/clickhouse/store/a7a/a7a4bb8f-.../202610_1_1_1/

This part is Wide: each column has its own .bin file and its own marks file, which is also why a query that reads one column of a wide table never touches the others. The small metadata files are still there, about the same size as in the one-row part.

	One order per INSERT	One million orders in one INSERT
Part type	Compact	Wide
Files	13	21
Column data (`.bin` files)	161 bytes	12.2 MiB
On disk (`du`)	56 KiB	12.3 MiB
Share that is column data	0.3%	99.3%
Disk per order	56 KiB	about 13 bytes

The overhead of a part is roughly fixed, so with a million rows behind it you stop noticing it: about 13 bytes per order. (My test data flatters this a bit, since status cycles through three values and compresses to almost nothing, but the per-part overhead is the same either way.)

So a part never changes: new data means a new part, and combining parts means writing a bigger one and deleting the old ones later. Each part is sorted by ORDER BY, but only inside itself, so two parts can both hold the same order_id, which is exactly how duplicates happen. Each part also belongs to one partition and carries its own index, marks and checksums, and a SELECT reads every active part its indexes don't rule out.

The part name 202610_1_1_1 encodes its history. The official docs for system.parts explain the format:

The part naming format is the following:
<partition_id>_<minimum_block_number>_<maximum_block_number>_<level>_<data_version>
partition_id - identifies the partition key

minimum_block_number - identifies the minimum block number in the part. ClickHouse always merges continuous blocks

maximum_block_number - identifies the maximum block number in the part

level - incremented by one with each additional merge on the part. A level of 0 indicates this is a new part that has not been merged. It is important to remember that all parts in ClickHouse are always immutable

data_version - optional value, incremented when a part is mutated (again, mutated data is always only written to a new part, since parts are immutable)

So 202610_1_1_1 is partition 202610, blocks 1 to 1, level 1. The level surprised me, since the docs say a new part starts at 0. It turns out ReplacingMergeTree runs its merge logic on every inserted block before writing it, so its new parts count as merged once already. The block number raised even more questions, because "block" means three different things in ClickHouse.

I followed both rabbit holes, together with everything else the server does inside one INSERT: splitting rows by partition, sorting, writing marks and the primary index, and the commit. That turned into a post of its own, Inside one INSERT, which is Part 3 of this series. You don't need it for the rest of this one.

2. Too many parts

"Why are we getting Too many parts errors?"

Every part costs files on disk, metadata in memory, and a future merge. If a partition piles up parts faster than merges can combine them, ClickHouse protects itself: at 1000 active parts in a partition (parts_to_delay_insert) it starts slowing INSERTs down on purpose, and at 3000 (parts_to_throw_insert) it rejects them with Too many parts. Those are the current defaults. A lot of older posts still quote 150 and 300, which were the defaults before 23.6.

The classic cause is many small INSERTs. To see it, I used a separate test table and sent it 50 single-row INSERTs concurrently, with merges stopped:

for i in $(seq 1 50); do
  docker exec ch clickhouse-client --async_insert=0 -q "INSERT INTO ins_test (id) VALUES ($i)" &
done; wait

With async_insert = 0, the 50 rows became 50 parts. Running the same loop with async_insert = 1 gave 7 parts for the same 50 rows. With asynchronous inserts, the server buffers incoming INSERTs and writes one part per flush. A flush happens after 200 ms (async_insert_busy_timeout_max_ms) or 10 MiB (async_insert_max_data_size). Since 26.2, async_insert is on by default.

Since async inserts are on by default now, I assumed they had already batched my three single-row INSERTs into orders. They hadn't. With wait_for_async_insert = 1 (the safe default) each INSERT waits until its data is flushed, so a client that sends INSERTs one after another never shares a buffer with anyone. Async inserts batch concurrent writers, like many app servers or Lambdas; a single sequential loop still has to batch on the client side.

3. Merges, and why nothing happened for six minutes

Background merges combine adjacent parts within one partition into a bigger part, using a streaming k-way merge of already-sorted data. Let's turn them back on:

SYSTEM START MERGES orders;

I expected a merge within seconds. Nothing happened. Thirty seconds later the three parts were still there, and they stayed separate for the next six minutes while I wondered whether I'd forgotten something.

The reason is in SimpleMergeSelector.h: the selector prefers to merge about five parts of similar size at once (base = 5), which keeps write amplification low, and relaxes that only gradually as parts get older or more numerous. Three tiny, fresh parts aren't worth it yet. That's the 26.9 behaviour, at least; the selector has its own table settings, and its rules change between releases.

So I inserted three more orders. ClickHouse records every part event in system.part_log:

SELECT event_time_microseconds AS t, event_type, part_name, merged_from, rows
FROM system.part_log
WHERE table = 'orders'
ORDER BY t;

┌──────────────────────────t─┬─event_type──────┬─part_name────┬─merged_from──────────────────────────────────────────┬─rows─┐
│ 2026-10-10 11:10:22.332555 │ NewPart         │ 202610_1_1_1 │ []                                                   │    1 │
│ 2026-10-10 11:10:22.489426 │ NewPart         │ 202610_2_2_1 │ []                                                   │    1 │
│ 2026-10-10 11:10:22.654394 │ NewPart         │ 202610_3_3_1 │ []                                                   │    1 │
│ 2026-10-10 11:16:09.673250 │ NewPart         │ 202610_4_4_1 │ []                                                   │    1 │
│ 2026-10-10 11:16:09.829828 │ NewPart         │ 202610_5_5_1 │ []                                                   │    1 │
│ 2026-10-10 11:16:09.830053 │ MergePartsStart │ 202610_1_5_2 │ ['202610_1_1_1','202610_2_2_1',...,'202610_5_5_1']   │    0 │
│ 2026-10-10 11:16:09.832240 │ MergeParts      │ 202610_1_5_2 │ ['202610_1_1_1','202610_2_2_1',...,'202610_5_5_1']   │    5 │
│ 2026-10-10 11:16:09.984342 │ NewPart         │ 202610_6_6_1 │ []                                                   │    1 │
└────────────────────────────┴─────────────────┴──────────────┴──────────────────────────────────────────────────────┴──────┘

The first three parts waited almost six minutes, but when the fifth part arrived, a merge of exactly five parts started 0.2 ms later. The result, 202610_1_5_2, covers blocks 1–5 at level 2, and the sixth part is left waiting for company.

Merged-away parts stay on disk as active = 0 for another 8 minutes (old_parts_lifetime), so queries already reading them can finish.

What I take from this: merges are eventual, and you can't predict when, or even whether, two given rows end up in the same part. The six minutes and the five parts are just what I saw in one run, not numbers to plan around. And merges never cross partitions, which matters a lot in the next section.

4. Updating the order: duplicates and FINAL

"Why do I still see the old value? I'm using ReplicatedReplacingMergeTree."

Order 1 gets paid. In a ReplacingMergeTree, an update is just another INSERT with a newer version:

INSERT INTO orders VALUES (1, 'paid', 10.00, now64());

SELECT * FROM orders WHERE order_id = 1;

┌─order_id─┬─status──┬─amount─┬──────────────updated_at─┐
│        1 │ created │     10 │ 2026-10-10 11:10:22.330 │
│        1 │ paid    │     10 │ 2026-10-10 11:16:26.581 │
└──────────┴─────────┴────────┴─────────────────────────┘

Two rows, because the old version is in one part and the new one in another, and nothing has merged them yet. You can read it correctly either way:

-- Let ClickHouse apply the merge logic at read time
SELECT * FROM orders FINAL WHERE order_id = 1;          -- paid

-- Or deduplicate yourself
SELECT order_id, argMax(status, updated_at) AS status
FROM orders WHERE order_id = 1 GROUP BY order_id;       -- paid

FINAL has a reputation for being slow, but much of that advice is old: today it runs in parallel and skips parts that can't contain duplicates (that's where those level-1 parts help). Its cost still depends on the table layout and the query, so for anything performance-sensitive I'd measure it against argMax.

The partition trap

Now the order ships, and the shipment is recorded in November:

INSERT INTO orders VALUES (1, 'shipped', 10.00, '2026-11-02 09:00:00');
OPTIMIZE TABLE orders FINAL;   -- force a merge in every partition

SELECT _partition_id, _part, * FROM orders WHERE order_id = 1;

┌─_partition_id─┬─_part────────┬─order_id─┬─status──┬──────────────updated_at─┐
│ 202610        │ 202610_1_7_3 │        1 │ paid    │ 2026-10-10 11:16:26.581 │
│ 202611        │ 202611_8_8_2 │        1 │ shipped │ 2026-11-02 09:00:00.000 │
└───────────────┴──────────────┴──────────┴─────────┴─────────────────────────┘

Even after a forced full merge the duplicate is still there, and it will stay there forever, because the two versions are in different partitions and merges never cross partitions. FINAL still returns the right answer, because by default it deduplicates across partitions at read time, on every query. But turn on the popular optimization for FINAL:

SELECT * FROM orders FINAL WHERE order_id = 1
SETTINGS do_not_merge_across_partitions_select_final = 1;

┌─order_id─┬─status──┬──────────────updated_at─┐
│        1 │ shipped │ 2026-11-02 09:00:00.000 │
│        1 │ paid    │ 2026-10-10 11:16:26.581 │
└──────────┴─────────┴─────────────────────────┘

That's two rows for one order, with no error and no warning. If I had to pick one thing from this whole post to remember, it would be this: in a ReplacingMergeTree, every version of a row has to land in the same partition. Partition by something that never changes for a key, such as toYYYYMM(created_at), or don't partition at all.

Back to the coffee break

The short answers are in the TL;DR at the top. If I had to boil the rest down for a colleague: batch your INSERTs on the client, because async inserts only help with many concurrent writers. Don't panic about a few small parts, since merges are lazy on purpose, and system.parts and system.part_log will show you what's going on. Pick a partition key that never changes for a row, and read with FINAL or argMax instead of scheduling OPTIMIZE ... FINAL to clean up duplicates. Merges save storage, but your queries have to be correct without them.

DEV Community: Maksim Grishchenko

Generating Kafka ClickHouse pipelines from a JSON sample

TL;DR

The flow: streams, raw, datalake

The migration, layer by layer

First attempt: a shell script

clickforge

Closing words

When ClickHouse merges change your query results

1. When merges change the answer

SummingMergeTree: zero rows disappear

AggregatingMergeTree: a plain column changes its value after a merge

CollapsingMergeTree: the order of parts matters

The rule for all of them

2. Deleting and updating in place

Back to the coffee break

Further reading

Follow one row: what ClickHouse does with your INSERT

TL;DR

Preparations

1. An INSERT creates a part

What a part is

2. Too many parts

3. Merges, and why nothing happened for six minutes

4. Updating the order: duplicates and FINAL

The partition trap

Back to the coffee break

Further reading