Rebalancing thousands of tenants across ClickHouse shards without zero-copy replication
After two years of manual ClickHouse shard migrations, we tried an experiment: cold data in S3 as pointer files. Moving 5 days of a tenant meant copying 7.7 KB instead of 13 MB.
On this page
- The problem: ClickHouse scales, but it doesn’t rebalance
- Options we ruled out
- The idea: every table has two homes
- Step 1: split big tenants, then place every split
- Step 2: age data from local disk into object storage
- Step 3: rebalance in seconds
- We rebuilt it in a lab: the numbers
- Three surprises worth knowing before you build this
- 1. ATTACH writes to object storage, so read-only access is not enough
- 2. ClickHouse will try to delete your shared objects
- 3. DROP PARTITION can start an endless retry loop
- Who deletes the data, then?
- Where AI helped
- What’s next: a shared metadata layer
- Trade-offs to know about
- FAQ
- Does ClickHouse rebalance shards automatically?
- Why not use zero-copy replication?
- Can ClickHouse use a read-only S3 bucket for MergeTree parts?
- How long does moving a tenant take?
- Is this only useful for multi-tenant systems?
ClickHouse scales well. Add a shard and it happily takes new data. What it does not do is move your existing data to the new shard. With a handful of big tables that is a minor annoyance. With thousands of tenants of very different sizes, it turned into two years of manual data migrations for us.
This post describes an experiment we ran to make rebalancing automatic, and its results were better than we hoped. It covers how we decide where each piece of a tenant lives, how data ages from local disk into object storage, and why moving most of it now means copying a few small pointer files. To make sure every step holds up, we also rebuilt the core of it from scratch in a lab on ClickHouse 26.10, measured it, and found three behaviours you need to know about before you try this yourself.
The problem: ClickHouse scales, but it doesn’t rebalance
In a sharded cluster, a Distributed table decides which shard receives each new row. Once a row is written, it stays on that shard forever. So when you add capacity:
- New data spreads across the new layout
- Old data stays exactly where it was
- Big tenants keep their shard busy for as long as their history exists
With thousands of tenants, sizes vary by orders of magnitude, and a few large tenants can fill a shard while a new server sits nearly empty. Fixing that by hand means, for every tenant you move: copy its partitions, verify them, switch reads over, delete the source, and hope nothing was written in between. We did that for two years. It was slow, risky, and it always seemed to happen at night.
Options we ruled out
- Zero-copy replication. Several servers share the same objects in S3, which sounds perfect. But the ClickHouse docs say plainly that it is “not ready for production”, and it has been disabled by default since version 22.8.
- SharedMergeTree. It solves this problem properly, but it is only available in ClickHouse Cloud.
INSERT ... SELECTfrom the old shard. It works, but it reads, re-sorts and re-compresses every row, loads both servers, and triggers merges on the receiving side. Fine for one tenant, painful for hundreds.
What we wanted was a way to move data that does not rewrite it.
The idea: every table has two homes
Every table stores data on two disks:
- Hot data lives on local disk. This is where inserts land and merges happen.
- Cold data, older than a few days, has finished merging. It moves to object storage in ClickHouse’s own part format, and each server keeps only small pointer files that say where the real files are.
That split is what makes rebalancing cheap. Most of a tenant’s data is cold, and moving cold data means moving pointers, not bytes. One requirement follows from it: data has to be cut into partitions that can move on their own. In our system every tenant has its own set of about ten tables, and each table is partitioned by day and a split number: PARTITION BY (toYYYYMMDD(ts), split_id). The split number is what lets big tenants spread out, as the next step shows.
Step 1: split big tenants, then place every split
Tenants are not equal, so the unit we place is not a tenant. It is a slice: one split of one tenant.
- A small tenant writes everything into split 0.
- A separate program watches each tenant’s data size and ingest rate, and raises its number of splits as it grows. New rows are spread across the splits.
- Existing partitions keep the split they were written with. Raising the split count changes where new data goes; nothing already written is rewritten.
Because every slice is placed on its own, a large tenant’s slices land on different servers, and its data spreads across the cluster without anyone deciding where.
Placement itself is computed, never decided by hand. The rules we care about:
- Deterministic. The same server layout always produces the same answer
- Minimal movement. Adding one shard should move only the slices that must move
- Capacity-aware. Bigger slices count for more, and bigger servers take more
A good starting point is weighted rendezvous hashing: every slice scores every shard, and the highest score wins. Adding a shard only steals the slices for which the new shard now scores highest, so most slices stay put. The core idea in a few lines of Python:
import hashlib, math
def score(slice_id: str, shard: str, weight: float) -> float: # slice_id like "tenant42/split3"
h = int(hashlib.sha256(f"{slice_id}:{shard}".encode()).hexdigest()[:16], 16)
u = (h + 0.5) / 2**64 # uniform in (0, 1), never 0 or 1
return -weight / math.log(u) # weighted rendezvous score
def place(slices, shards): # shards: {name: capacity_weight}
return {x: max(shards, key=lambda s: score(x, s, shards[s])) for x in slices}
def moves(old_layout, new_layout):
return {x: (old_layout[x], s) for x, s in new_layout.items() if old_layout.get(x) != s}
A quick test of exactly this code: 1,000 tenants, where 700 have one split, 200 have two and 100 have eight, for 1,900 slices on 3 equal shards. Then a 4th shard is added. 466 slices moved (24.5%), and every one of them moved to the new shard. Nobody was shuffled between the old shards, and each eight-split tenant ended up spread over all four servers. Give the new shard double weight and it takes 755 of the 1,900.
Our experimental placement program goes further: it accounts for slice size, keeps shards under a target fill level and keeps a tenant’s splits apart. But the output is the same kind of thing: a list of “move this slice from shard A to shard B”. Rebalancing only runs when capacity changes, so most days that list is empty.
Step 2: age data from local disk into object storage
Once a partition is a few days old it no longer changes, so it ships to object storage:
- Snapshot it.
ALTER TABLE ... FREEZE PARTITIONcreates hard links to the part files under the server’sshadow/directory. It is nearly instant and copies nothing. - Upload the files as they are. A separate uploader, with its own credentials, copies every file of every part unchanged. ClickHouse itself gets an S3 policy that can read and create objects, but never delete them. More on why below. The bucket holds valid ClickHouse parts with their normal file names, not an export format.
- Write pointer files. The server reads cold data through an S3 disk that keeps its metadata locally: one tiny text file per data file, naming the object and its size. On 26.10 a pointer file looks like this:
5
1 1141335
1141335 parts/events/3-20261101_3_20_1/value.bin
0
1
That is the format version, the number of objects and total size, then one line per object (size and key), a reference count and a read-only flag. It is an internal format, so pin your version and test every upgrade.
- Swap the cold copy in. Load the pointer parts into a staging table, then
ALTER TABLE events REPLACE PARTITION ... FROM staging. The local copy is replaced by the cold one in a single step, so queries never see the partition missing.
Step 3: rebalance in seconds
When the placement program says “move this slice from A to B”:
Cold partitions (most of the data):
- Copy the tenant’s pointer files from A to B. They are tiny.
- Load them on B through a staging table and swap them in with
REPLACE PARTITION. Nothing is downloaded; the objects are already in S3.
Hot partitions (only the last few days):
FREEZEthem on A for a consistent snapshot, thenrsyncthe part directories into B’sdetached/directory.ALTER TABLE ... ATTACH PARTITIONon B, which is instant because the files are already there.
Then switch and clean up:
- Switch the tenant’s routing to B.
- Remove the tenant from A:
DETACH PARTITION, then delete the detached pointer folders. The S3 objects stay, because B now uses them. ClickHouse never runsDROP PARTITIONin this design; see the third surprise below for why.
We rebuilt it in a lab: the numbers
To check every step, we rebuilt the core of this design from scratch: two ClickHouse 26.10 servers and an S3-compatible object store, one table partitioned by tenant and day, and one tenant with 2 million rows over 8 days. Five days were cold, three were hot. Then we moved the tenant from server 1 to server 2.
| What moved | Real data size | What was actually copied |
|---|---|---|
| 5 cold days | 13.0 MB of parts in S3 | 7.7 KB of pointer files (120 files) |
| 3 hot days | 7.1 MB on local disk | 7.1 MB, copied as-is |
Uploading the five cold partitions took 176 ms. Loading them back took about 100 ms. After the move, row counts, sums and distinct-user counts on server 2 matched the original exactly. The ratio is the point: the cold side of a move costs the same few kilobytes whether the tenant has 13 MB or 13 TB in object storage.
Three surprises worth knowing before you build this
1. ATTACH writes to object storage, so read-only access is not enough
Our first idea was to give ClickHouse read-only credentials, so it could never damage the shared copy. On 26.10 that breaks ATTACH PARTITION and REPLACE PARTITION. Both rename the part, and while doing so ClickHouse writes one new 28-byte object (invalidated_system_columns.txt). With read-only credentials the query fails with Access Denied. Uploading that file in advance didn’t help: ClickHouse writes it again anyway.
So the rule we settled on is read and create, but never delete:
{
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:PutObject", "s3:ListBucket"],
"Resource": ["arn:aws:s3:::cold-bucket", "arn:aws:s3:::cold-bucket/*"]
}
There is no s3:DeleteObject anywhere in the policy. ClickHouse writes its new objects under random names, so it cannot overwrite the uploaded parts, and it cannot remove anything. In the lab, attach and replace worked normally with this policy.
If you need strictly read-only access, there is another way: load parts without renaming them. DETACH TABLE staging, put the pointer folders straight into the table’s data directory, ATTACH TABLE staging. That needs no writes at all, at the cost of a slightly more manual process.
Older ClickHouse versions may not write this file. Test on the exact version you run.
2. ClickHouse will try to delete your shared objects
When an ATTACH failed halfway, ClickHouse cleaned up by trying to delete every object the pointer files referenced: 1,620 delete attempts in our lab. Dropping a partition did the same. The read-only flag inside the pointer files did not stop it. The only thing that kept the data safe was the S3 policy that refused deletes.
That is why “never delete” is not optional: ClickHouse must never hold delete permission on shared cold objects. Turn on bucket versioning as a second safety net; it also protects against an accidental overwrite.
3. DROP PARTITION can start an endless retry loop
In the lab, we dropped the moved tenant on server 1 to see what would happen. Its cleanup thread kept retrying the refused deletes, roughly 100 requests a second, until the server restarted. It doesn’t break anything, but it floods logs and wastes requests.
That is why ClickHouse never runs DROP PARTITION on cold data in our design. Our app does it in two steps instead:
ALTER TABLE events DETACH PARTITION (3, 20261101);
-- then the app deletes that partition's folder from detached/ with a normal file removal
Detaching moves the pointer folder into detached/, and removing that folder from the filesystem never touches S3. In the lab this produced no delete attempts at all.
Who deletes the data, then?
ClickHouse never deletes cold objects, so our app owns deletion completely. There are two cases, and the difference between them is the most important rule in the whole system.
Moving a slice (rebalancing): detach on the old server and remove its pointer files. Never touch S3. The new server reads the same objects.
Removing data (retention, or a tenant leaving): this is the only time objects are deleted.
DETACH PARTITIONon every server that holds the partition.- Remove the detached pointer folders.
- Delete the objects from S3, using the app’s own credentials (the only ones with delete permission).
Step 3 only runs once no server references those objects any more. The app can check that, because it already knows where every partition lives: that is the placement map from step 1. As a safety net, an occasional sweep compares every pointer file on every server with the bucket listing and reports objects that nobody references.
Where AI helped
We had the outline of this design for a long time. What we never had was the time to build it next to everything else. AI changed that. It helped us write the placement logic and its tests, script the freeze, upload and attach steps, and work through the edge cases: what happens if a move stops halfway, how to check a part arrived intact, how to resume.
It didn’t come up with the idea; we did. It shortened the distance between a whiteboard and a working experiment, which is the part that had stalled for two years.
What’s next: a shared metadata layer
The experiment worked better than we expected, and it also showed us where the remaining friction is. Pointer files still live on each server’s local disk, so every move is still a small file copy, and every server keeps its own view of which parts it owns.
So we are now working on the next step: a shared metadata layer, similar in spirit to ClickHouse Keeper, that knows which parts exist in object storage and which server serves them. With it, attaching or detaching a part becomes a metadata change instead of moving files around, and we want to get there with as few changes to ClickHouse itself as possible. The goal is the kind of features SharedMergeTree offers in ClickHouse Cloud, along with the performance, on self-managed ClickHouse.
It is early work. We will write about it once there is something to measure.
Trade-offs to know about
- Cold data is immutable. No merges, no mutations. Late-arriving rows land in the hot tier; changing old rows means rewriting the partition.
- Pointer files are an internal format. They can change between ClickHouse versions. Pin your version and run a lab test like ours before every upgrade.
- Deletion is yours. ClickHouse never deletes cold objects, so your app must, and it must know when no server needs them any more.
- Object storage is slower than NVMe. A filesystem cache in front of the cold disk keeps repeated queries fast.
- Verify every move. Compare row counts and a few aggregates on both servers before removing the source, and keep each part’s
checksums.txt.
FAQ
Does ClickHouse rebalance shards automatically?
Not in self-managed ClickHouse. Adding a shard changes where new rows go, but existing data stays where it was written. You have to move it yourself.
Why not use zero-copy replication?
The ClickHouse docs say it is not ready for production, and it has been disabled by default since 22.8. Sharing objects between servers is the right idea, but we wanted ClickHouse to read and add shared data, never delete it.
Can ClickHouse use a read-only S3 bucket for MergeTree parts?
For reading, yes. But on 26.10, ATTACH PARTITION and REPLACE PARTITION write a small new file, so strictly read-only credentials make them fail. The safe middle ground is a policy that allows reading and creating objects but never deleting them. If you must stay strictly read-only, load parts by re-attaching a staging table instead.
How long does moving a tenant take?
The cold part of a move copies a few kilobytes of pointer files, no matter how much data is in object storage. Only the last few days of hot data are actually copied, so a move takes minutes at most.
Is this only useful for multi-tenant systems?
No. Any cluster where data has a natural owner (a tenant, a customer, a project) and grows unevenly can use the same pattern.