Build a Multi-Node Cluster with Citus

You’ve mastered single-node PostgreSQL — indexes, transactions, EXPLAIN. But what happens when your hot table blows past a terabyte, or your analytics query drags because one CPU is doing all the work? You could buy a bigger machine, but that hits a ceiling fast and gets expensive. The answer many production teams reach for is horizontal scaling: spread the data and the query load across many inexpensive nodes. And the most accessible way to do that with PostgreSQL is Citus, an extension that turns a cluster of Postgres instances into one logically distributed database. This lesson walks you through building a multi-node Citus cluster from scratch, distributing tables, and running queries that span all your machines.

The problem this lesson solves

Scaling a relational database vertically has hard limits. At some point, you can’t rent a bigger instance, and even if you can, the cost per GB grows super-linearly. Meanwhile, demand for analytics and real-time features keeps climbing. Citus solves this by giving you scale-out: instead of one database doing everything, you have a coordinator node that plans queries and a set of worker nodes that store shards of your data. Your application still talks to one PostgreSQL endpoint, so you don’t rewrite your queries or learn a new query language. The problem this lesson solves is the operational gap: how do you actually stand up that coordinator and those workers, connect them, load data, and verify that queries are being distributed — rather than just reading about it?

Core concept / mental model

Think of Citus as a conductor and an orchestra. The coordinator (the conductor) doesn’t store much of your data itself; it keeps metadata about where every shard lives and routes queries to the right players. The workers (the orchestra) hold the actual data, each responsible for a subset of rows. When you run a query, the coordinator breaks it into pieces, sends them to the workers, and then merges the results — all behind the scenes.

A key mental model is the shard. A shard is simply a horizontal slice of a table. Citus splits a distributed table into N shards, each stored as a regular PostgreSQL table on one of the workers. The default shard count is 32 (per table), and Citus places them across workers for balance. When you distribute a table, you choose a distribution column — this is the column whose value decides which shard a row lands in. For example, if your app is multi-tenant, you’d pick tenant_id. This choice is critical: queries that filter on that column can be answered by a single worker (fast), while queries that don’t will scatter across all workers (slower but still parallel).

Think of the distribution column as the shard key, similar to the partition key in PostgreSQL declarative partitioning, but with a cluster-wide effect. Citus uses a hash of that column to map rows to shards, and each shard lives on one worker. Your job is to pick the right column by imagining your most frequent and most important queries — they should all filter on that column.

How it works step by step

Building a Citus cluster involves a few distinct phases: preparation, configuration, setup, and verification. Here’s the cause-and-effect sequence you need to internalize:

1. Provision nodes — You need at least one coordinator and two workers (Citus recommends at least 2 workers for real scale). Each node is a separate PostgreSQL instance, possibly on separate VMs or containers.

2. Install Citus on every node — The extension must be installed and enabled on every node, including the coordinator. Without it, the nodes can’t understand the cluster metadata functions.

3. Configure shared library — In postgresql.conf, set shared_preload_libraries = 'citus'. This loads the extension into shared memory at server start. Then restart PostgreSQL on each node.

4. Start the coordinator and workers — Start each PostgreSQL instance normally.

5. Add workers to the coordinator — On the coordinator, run SELECT * from citus_add_node('worker-host', port);. This registers each worker, and Citus marks the cluster as active.

6. Verify the cluster — Run SELECT * FROM citus_get_active_worker_nodes(); to confirm the workers are visible.

7. Create and distribute tables — Use CREATE TABLE as usual, then call select create_distributed_table('table_name', 'distribution_column');. This splits the table into 32 shards (by default) and places them across workers.

8. Load data and query — Insert rows normally; Citus routes them to the right shard based on the hash of the distribution column. Run queries against the coordinator, and it will push down what it can to the workers.

The key is that after step 6, your cluster is live. From then on, you write standard SQL. Citus handles the sharding and parallel query execution automatically.

Pro tip: Use a dedicated network or at least a private subnet for the nodes, and set citus.node_conninfo to a valid sslmode or require to encrypt connections. In production, never expose worker ports publicly.

Hands-on walkthrough

Let’s actually build a two-worker cluster. I’ll assume you have three Debian/Ubuntu servers (or containers) named coordinator, worker1, and worker2, each running PostgreSQL 15 (or later). The steps are identical for PostgreSQL 16/17.

Step 1: Configure PostgreSQL on each node

Edit postgresql.conf (typically at /etc/postgresql/15/main/postgresql.conf) on every node to add:

shared_preload_libraries = 'citus'

Then restart PostgreSQL:

sudo systemctl restart postgresql

Step 2: Install and enable Citus

On each node, install the extension (if not already done with the package) and then enable it in your application database:

sudo apt-get install postgresql-15-citus

Then connect to the database you’ll use (call it appdb) and run:

CREATE EXTENSION citus;

Verify it’s there:

SELECT * FROM pg_extension WHERE extname = 'citus';

Step 3: Add worker nodes from the coordinator

On the coordinator, register each worker:

-- Run on coordinator
SELECT * from citus_add_node('worker1', 5432);
SELECT * from citus_add_node('worker2', 5432);

If you use containers, use the container names or IP addresses. Check the cluster:

SELECT * FROM citus_get_active_worker_nodes();

Expected output (assuming hostnames worker1 and worker2):

 nodename | nodeport
----------+----------
 worker1  |     5432
 worker2  |     5432
(2 rows)

Step 4: Create and distribute a table

Now create a table that you want to distribute. For a multi-tenant SaaS, a natural choice is events:

CREATE TABLE events (
    tenant_id bigint,
    event_id bigint,
    payload jsonb,
    created_at timestamptz
);

SELECT create_distributed_table('events', 'tenant_id');

This splits events into 32 shards by default, spread across worker1 and worker2.

Step 5: Load data and query

Insert rows as you normally would — Citus sends each row to the right shard:

INSERT INTO events (tenant_id, event_id, payload, created_at)
SELECT g, g, jsonb_build_object('n', g), now() FROM generate_series(1, 100000) g;

Now run a query and verify it uses parallelism:

EXPLAIN ANALYZE SELECT count(*) FROM events WHERE tenant_id = 42;

Because you filtered on the distribution column, the coordinator should push the query down to a single shard. The explain plan will show Task nodes and possibly a Custom Scan (Citus).

Try a query without the distribution column filter:

EXPLAIN ANALYZE SELECT tenant_id, count(*) FROM events GROUP BY tenant_id;

This will fan out to all workers, each counting its local shards, and the coordinator merges results. You’ll see multiple Task entries — that’s parallel execution working.

Compare options / when to choose what

Citus isn’t the only way to scale PostgreSQL. Here’s how it stacks against common alternatives:

When to choose Citus: You have a clear distribution column (e.g., tenant_id, user_id), you need horizontal write scaling, and you can tolerate some cross-shard query costs. When not: Your queries are heavily relational with many JOINs across unrelated entities, or your dataset fits comfortable on a single high-end machine.

Variations:

• Citus local tables: Keep reference tables (like product codes) as local on the coordinator for fast joins.
• Reference tables: Use create_reference_table for small tables that need to be present on every worker to avoid cross-shard joins.
• Shard rebalancing: Use rebalance_table_shards() to redistribute shards when adding new workers.

Troubleshooting & edge cases

Error: extension "citus" is not available

Cause: The Citus package isn’t installed or not in the default library path. On Ubuntu, install postgresql-15-citus (matching your PG version) and retry. If you compiled from source, check dynamic_library_path.

Error: could not connect to worker on citus_add_node

Cause: The worker’s listen_addresses doesn’t include the coordinator’s IP, or pg_hba.conf rejects the connection. Set listen_addresses = '*' on workers and add a line like host all all coord_ip/32 trust. For production, use scram-sha-256.

Query runs on coordinator only

If EXPLAIN shows no Custom Scan (Citus) and no tasks, the table wasn’t actually distributed. Check SELECT * FROM citus_tables; — if your table is listed as local, run create_distributed_table again. Also, make sure you’re connected to the coordinator, not a worker.

High query latency / no parallelism

Make sure your distribution column matches how you filter. Queries that don’t filter on it will scatter, but they still should show parallelism. If you see sequential scans on the coordinator, check whether citus.enable_parallel_query is set in postgresql.conf.

Data inconsistency after adding a worker

New workers won’t automatically get shards until you run rebalance_table_shards(). This is normal — run the function to rebalance.

Edge case: small reference tables

If you join a large distributed table with a small lookup table (like countries), you’ll get a cross-shard join unless you make the lookup a reference table via create_reference_table('countries'). This duplicates the small table on every worker.

What you learned & what's next

You now understand the problem of single-node scaling, the mental model of coordinator and workers, and the step-by-step mechanics of building a Citus cluster. You’ve completed a hands-on exercise: installing Citus, adding workers, distributing a table, and inspecting query plans. You can compare Citus to alternatives and troubleshoot common setup failures. These are the core objectives of this lesson.

• You can explain the core idea behind building a multi-node Citus cluster.
• You can complete a practical exercise to build one.
• You know when Citus is the right tool and when it isn’t.

Next in the PostgreSQL Tutorial, you’ll likely explore monitoring and performance tuning for such a cluster — or perhaps dive into distributed transactions and consistency guarantees. Great job making your Postgres horizontally scalable!