Citus doesn’t actually shard PostgreSQL tables; it distributes them by replicating shard metadata and then distributing the actual data across multiple PostgreSQL nodes.
Let’s see this in action. Imagine we have a single PostgreSQL instance and we want to distribute our events table across multiple nodes.
First, we need to install the Citus extension. On a standard PostgreSQL installation, this would look something like:
CREATE EXTENSION citus;
Now, let’s create our events table. This is a regular PostgreSQL table at this point.
CREATE TABLE events (
event_id BIGSERIAL PRIMARY KEY,
event_type VARCHAR(50),
event_time TIMESTAMPTZ DEFAULT NOW(),
user_id BIGINT,
payload JSONB
);
To distribute this table, we need to tell Citus how to shard it. The most common way is by using a distribution column. For events, user_id is a good candidate because we often query events by user.
SELECT citus_dist_table('events', 'user_id');
After running this command, Citus takes over. It doesn’t magically move data. Instead, it creates metadata that tracks where each shard of the events table should live. The actual data is then distributed across the worker nodes you’ve configured. When you insert a new event, Citus determines which worker node the user_id belongs to and directs the insert there. When you query by user_id, Citus knows exactly which worker node(s) to ask for the data.
Here’s what the internal state looks like after distribution. Citus maintains a pg_dist_logical_replication_conf table on the coordinator node. This table stores information about the distributed tables, their distribution columns, and the number of shards.
-- On the coordinator node:
SELECT logicalrelid::regclass, colname, shardid, shardcount, placementid
FROM pg_dist_shard JOIN pg_dist_placement ON shardid = id
JOIN pg_dist_logical_replication_conf ON shardid = conf.objid
ORDER BY shardid;
This query would show you that the events table is now logically sharded. You’ll see entries for each shard of the events table, indicating which placementid (which worker node and its associated shard copy) is responsible for that shard.
The problem Citus solves is scaling out PostgreSQL horizontally. As your data grows and your query load increases, a single PostgreSQL instance can become a bottleneck. Sharding allows you to spread the data and the query load across multiple machines. Citus makes this process manageable by handling the complexity of distributed query planning and data placement.
The magic behind Citus is its query router. When you execute a query against a distributed table, the coordinator node receives the query. It then consults its metadata to determine which worker nodes hold the relevant data shards. For queries that involve joins between distributed tables, Citus can perform distributed joins, sending parts of the query to worker nodes and aggregating results on the coordinator. For queries that only touch data on a single shard (e.g., WHERE user_id = 123), Citus can route the query directly to the specific worker node holding that shard, achieving near-local query performance.
One thing most people don’t realize is how Citus handles distributed transactions. While Citus aims for strong consistency, the underlying distributed nature means that certain operations, especially those involving multiple shards across different workers, are subject to the CAP theorem’s trade-offs. Citus uses a two-phase commit (2PC) protocol for distributed transactions to ensure atomicity across shards. However, if a worker node fails during a transaction, the transaction can be left in an uncertain state, requiring manual intervention or careful configuration of recovery mechanisms to resolve.
The next concept you’ll encounter is how Citus handles distributed joins and how to optimize them.