Distributed Table and Cluster Function Rewrites for Provider Migration

Migrating between cluster providers means rewriting every Distributed table and every cluster() call in the system, because both encode the exact topology of the fleet you're leaving. Get a cluster name or a replica address wrong, and queries don't always fail loudly. Sometimes they just quietly return less data than they should, and nobody notices until a report looks off three weeks later.

The core issue is that a Distributed table is not storage; it's a routing layer. It's a routing layer. It takes a query, fans it out to shards backed by ReplicatedMergeTree tables, and stitches the results back together. That routing depends on a cluster name string matching an entry in the server's remote_servers config, and that config is entirely specific to whoever built your fleet. None of it travels with the schema. So before touching a single line of DDL, the actual work starts with finding out what's plugged into that cluster name.

Taking a full inventory of distributed-layer dependencies before touching any DDL

Treat this like a dependency audit, not a config change. The goal is a list, exhaustive, of every object in the system that references a cluster name or a provider-specific host address. Missing one turns it into the thing that breaks two weeks after everyone's stopped watching the dashboards closely.

A few places to check, in order of how easy they are to miss:

system.tables, filtered on engine = 'Distributed'. The engine_full column shows the full engine definition including the arguments as written. system.query_log, searched for query text referencing cluster invocations. Ad-hoc queries and application calls never appear in table DDL at all, so this catches those. Materialized View definitions. Query system.tables where engine = 'MaterializedView' and inspect the stored definition. If an MV reads from or writes through a Distributed table, it inherits the dependency, silently. Application code and ORM layers. cluster() calls buried in application queries are invisible to ClickHouse's own system tables. This means an actual source-code grep is required. Any other objects that reference a Distributed table as a backing source.

For each hit, record the cluster name as written, the local table and database it points to, and whether it sits on the read path, the write path, or both. Flag any Materialized Views downstream of it too.

If a partial inventory catches the Distributed table but misses the MV feeding it, the migration will silently drop or duplicate data during the cutover. A common pattern chains a source table into an MV, which writes into a Distributed table, which lands in a destination table. A partial inventory that catches the Distributed table but misses the MV feeding it will leave the write path broken while the read path looks completely fine. Reads keep working off old data. Writes go nowhere. That's a rough one to catch in a smoke test, because a quick test may only verify reads and not writes.

The output of this phase should be a map: which cluster names exist, which objects touch each one, and which side of the read/write line they're on.

What exactly must change in a Distributed table definition when the cluster changes

The Distributed engine takes four arguments: ENGINE = Distributed('cluster_name', 'database', 'local_table', [sharding_key]). Of those, the cluster name is the one that has to change, because it's a lookup key into remote_servers on the node executing the query, and that lookup key is provider-assigned. It was never meant to be portable.

Database and local table names usually stay put, assuming the schema on the destination mirrors the source. If naming conventions differ over there, those change too, but that's the easy part. The sharding key expression is where it gets subtle: if the shard count on the destination matches the source, the key can stay as-is. If it doesn't, the same expression may distribute data unevenly across a different number of buckets. A sharding expression designed for one shard count won't distribute data evenly across a different number of shards. Review this every time shard count changes, full stop.

The remote_servers.xml rewrite carries its own set of rules:

Every <host> entry needs to be a hostname or FQDN reachable from both the source and destination clusters during the cutover window.
Source and destination cluster stanzas need to coexist in the config, not replace one another, at least until cutover finishes. In-flight queries and MV writes will fail if one side vanishes from the config mid-migration.
If traffic routes through a provider load balancer, the replica stanza looks different: a single host entry, a port, <secure>1</secure>, and username and password fields. That's a different shape from direct replica addressing, and mixing the two patterns by accident is a common source of confusing connection errors.
Ports differ across providers too. Native protocol traffic often runs on 9000, but TLS variants use 9440. Whatever the destination expects, the config has to match it exactly.

There's also the ZooKeeper (or Keeper) path on ReplicatedMergeTree tables, something like /clickhouse/tables/{shard}/table_name. The {shard} and {replica} macros expand based on each node's local macros config, so confirm those are set correctly on every destination node before assuming replication will just form on its own. It won't, if the macros are wrong.

The ON CLUSTER clause in CREATE TABLE ... ON CLUSTER 'source' DDL determines which nodes actually execute that statement. On the destination side, that clause needs the destination cluster name. Easy to miss, since it's just one string in a larger block of DDL, and everything else in that CREATE statement might be copy-pasted correctly.

What doesn't need to change: column definitions, ORDER BY, PARTITION BY, TTL rules, codec settings. Schema is portable. Topology isn't. Keep those two facts separate in your head while doing this work, because it's easy to start second-guessing schema decisions that were never actually part of the problem.

Rewriting cluster() function calls and the queries that contain them

The cluster('cluster_name', database, table) table function does roughly the same job as a Distributed table, sending a query out to every shard in the named cluster and merging what comes back, but it's written inline in SQL rather than defined once in DDL. That makes it more scattered and, frankly, easier to miss.

It also fails differently. A Distributed table pointing at a cluster name that's gone missing from config and a cluster() call with a bad cluster name can each fail in distinct ways that are worth understanding before testing begins. Different failure mode, worth knowing going in, because it changes how you'll notice the problem during testing.

Where do these calls hide? Ad-hoc entries in query_log, saved dashboard queries, BI tool connections, application code that builds SQL strings dynamically, and any structured pipeline definitions in use. That last category is at least systematic, since pipeline configs can usually be diffed file by file. Application code can't. That one needs an actual grep across the codebase.

The rewrite itself is mechanical: swap the old cluster name string for the new one. That swap has to happen after the destination cluster name is live in remote_servers, not before. That swap has to happen after the destination cluster name is live in remote_servers, not before. Pushing the query change ahead of the config change makes every one of those queries start failing immediately.

One function deserves separate mention: clusterAllReplicas(). Unlike cluster(), which hits one replica per shard, this one queries every replica. If replica count changes on the destination cluster, which is common when moving providers, the number of rows coming back changes too. Any query relying on that cardinality for deduplication logic downstream will start behaving differently, quietly, with no error.

And if the destination cluster runs on a newer distributed query planner, that's worth a separate pass. Planners that change how joins get distributed across shards, or how IN subqueries get pushed down, can shift result behavior even when the SQL text is identical. Don't assume identical output just because the query string didn't change. Running it both ways allows comparison.

How Materialized Views propagate the distributed dependency and must be migrated in order

Materialized Views are where migrations quietly fall apart, because the MV is the coupling point between a local table and a Distributed table, and order matters enormously here.

The typical pattern looks like this: a source local table takes all incoming INSERTs. An MV reads from that source table and writes into a Distributed table pointing at the destination. The destination's local table receives the routed rows. Three objects, one dependency chain, and every link in that chain has to exist and be internally consistent before any INSERT hits the source table.

Build in this order:

Create destination local tables first, using ON CLUSTER with the destination cluster name.
Deploy the updated remote_servers.xml with the destination stanza added alongside the source, not replacing it.
Create or update the Distributed table pointing at the destination.
Create or update the MV last, since it references objects that need to already exist.

Flipping that order creates a window where the MV tries writing into a Distributed table whose destination cluster isn't registered yet. That's not a subtle bug. It's a hard failure, but only during a narrow window that's easy to miss if the deploy happens fast.

CREATE MATERIALIZED VIEW ... POPULATE has an atomicity guarantee on the local insert path, when the relevant setting is enabled. But that guarantee doesn't stretch to rows arriving through a distributed write path. During the transition window, anything inserted via the old distributed route sits outside that atomicity boundary. That's a gap for whoever's running the cutover to document explicitly, since it's not the kind of thing that shows up in a quick test.

If MVs chain together, with one MV writing into a Distributed table that feeds another MV, map that whole chain before starting anything. Migrate it in dependency order, ensuring each target exists before the object that writes to it is rebuilt. Trying to migrate that chain top-down is how you end up with an MV firing into a target that doesn't exist yet.

Once the pieces are in place, confirm the MV is actually firing. Checking query_log can help confirm that writes are landing in the destination as expected. Don't take it on faith just because the CREATE statement ran clean.

The async vs. sync INSERT mode decision and its effect on migration safety

Distributed INSERTs run in one of two modes, and which one is active during cutover has real consequences for whether data actually lands where it's supposed to.

Async mode, the default, writes incoming data to a local spool directory on the initiating node. A background thread then forwards that data to the shards on its own schedule. The INSERT call returns success the moment the write hits the local spool, well before the data reaches its destination.

That's fine under normal operation. During a migration, it's a liability. If the destination cluster is unreachable when that background thread tries to forward its batch, data piles up in the spool. If the spool fills, or if the node gets decommissioned before it finishes draining, that data is gone. No error, no retry, no easy way to know it happened until someone notices missing rows.

Sync mode blocks the INSERT until the shards confirm the write. Slower, but there's no spool to accumulate and nothing to lose if a node disappears mid-migration.

Two related settings shape how the async path behaves under stress: batching individual inserts together for throughput, and controlling how failures on one shard affect the rest of the batch. Both are throughput optimizations, and both work against you during a cutover, since they're designed to keep things moving even when something downstream is unhealthy, which is exactly the situation you're trying to catch.

The safer move: switch the Distributed table pointing to the destination into sync mode for the duration of cutover. Confirm the destination is stable, then switch back to async for normal throughput once the window closes. Document that window clearly, with exact start and end times, so anyone debugging a gap later knows exactly when the mode was different.

And check the spool directory size on every node, during cutover and for a while after. A shrinking spool means things are draining the way they should. A flat or growing spool means the destination isn't keeping up, or isn't reachable, and cutover isn't actually done yet no matter what the dashboards say.

Schema changes that must accompany the topology rewrite

Not every schema change runs on the same clock, and lumping them all into one migration script is how timelines quietly slip. Some DDL is instant. Some triggers a background mutation that runs for minutes or hours. Knowing which is which changes how the whole cutover gets scheduled.

Adding a column is usually instant: metadata-only, no data rewrite. Dropping a column, or changing a column's type when the underlying data actually needs to be rewritten, along with bulk UPDATE and DELETE mutations, run in the background and can take a while depending on data volume. Running any of those while a Distributed table is actively writing into the affected parts risks inconsistency between replicas, since the mutation and the incoming writes are touching the same data at the same time.

A safer pattern for something like a column type change:

Add the new column with a default expression. Instant, metadata only.
Materialize the new column as a background mutation, and track its progress through system.mutations.
Only rename columns to swap the new one in once that mutation shows complete on every single replica, not just the one running the script.

Don't fold steps two and three into a single script. The mutation runs asynchronously, and no script can sit around waiting for it to finish reliably across every node.

ALTER statements wait for replica confirmation, and a network hiccup during the migration window (which is more likely than usual, given everything else moving at once) can leave some replicas on the old schema while others move ahead. Check system.mutations on every replica individually, since any one of them might have been the node that happened to run the ALTER.

As for what's safe to push to later: anything that's purely cosmetic, index tuning, TTL adjustments, minor codec changes, can generally wait until after cutover is confirmed stable. The topology rewrite is the load-bearing work. Schema polish can happen once the dust settles and the spool directories are all sitting quiet.