Citus

What Citus does

Citus introduces a coordinator-worker architecture on top of PostgreSQL. The coordinator node holds metadata about how tables are sharded and routes queries to the correct worker nodes. Worker nodes store the actual data shards and execute queries locally. Applications connect to the coordinator using standard PostgreSQL clients — the distribution layer is transparent.

Tables in a Citus cluster fall into three categories:

• Distributed tables are sharded by a distribution column. Each row is assigned to a shard based on the hash of its distribution column value. Queries that include the distribution column in their WHERE clause are routed to a single shard; broader queries are parallelized across all shards.
• Reference tables are small lookup tables that are replicated in full to every worker node. This allows them to be joined efficiently with any distributed table without cross-node traffic.
• Local tables are ordinary PostgreSQL tables that live only on the coordinator. They are useful for metadata, configuration, or administrative data that does not need to be distributed.

The coordinator rewrites incoming SQL into per-shard queries, pushes computation down to workers wherever possible, and merges results. For queries scoped to a single tenant or distribution key, the entire query runs on one worker with no coordination overhead.

When to use Citus

Citus is not needed for every PostgreSQL deployment — and I should be direct about this, because the allure of horizontal scaling has a way of arriving before the need for it does. Most databases perform well on a single node with proper indexing and query optimization. Citus becomes relevant when:

• Multi-tenant SaaS applications — shard by tenant_id so each tenant's data lives on a single shard, queries are fast, and tenant isolation is natural
• Real-time analytics dashboards — parallelize aggregation queries across workers to maintain sub-second response times as data volume grows
• Time-series data at scale — distribute high-ingest event or IoT data across nodes when a single PostgreSQL instance cannot keep up with write throughput
• Data that outgrows a single node — when your dataset exceeds what fits in memory on one server and disk I/O becomes the bottleneck
• Read and write scaling — unlike read replicas which only scale reads, Citus scales both reads and writes across the cluster

Installation and setup

Citus must be loaded via shared_preload_libraries at server startup, and it must be the first entry in the list. This is a hard requirement — PostgreSQL will not start if Citus is not listed first. A restart is required when enabling Citus for the first time. Allow me to save you the debugging session: if your server refuses to start after enabling Citus, check the ordering.

-- 1. Add to postgresql.conf (requires restart)
-- Citus must be first in the list
shared_preload_libraries = 'citus'

-- 2. Restart PostgreSQL, then create the extension
CREATE EXTENSION citus;

-- 3. Verify
SELECT citus_version();

For a multi-node cluster, you also need to register worker nodes from the coordinator.

-- On the coordinator node, register worker nodes
SELECT citus_set_coordinator_host('coordinator-host', 5432);
SELECT citus_add_node('worker-1', 5432);
SELECT citus_add_node('worker-2', 5432);

-- Verify the cluster
SELECT * FROM citus_get_active_worker_nodes();

Citus is available as a package from the Citus repository for Debian, Ubuntu, and RHEL-based distributions. It is also available as a Docker image. For single-node development, you can run the coordinator and workers on the same machine, or use Citus in single-node mode (Citus 11+) without any workers at all.

Key concepts and examples

Distributing a table

Create the table with standard DDL, then call create_distributed_table() with the distribution column. Citus splits the table into 32 shards by default (configurable via citus.shard_count) and distributes them across workers.

-- Create a table normally
CREATE TABLE events (
  tenant_id  bigint NOT NULL,
  event_id   bigint NOT NULL,
  event_type text NOT NULL,
  payload    jsonb,
  created_at timestamptz DEFAULT now(),
  PRIMARY KEY (tenant_id, event_id)
);

-- Distribute it by tenant_id (32 shards by default)
SELECT create_distributed_table('events', 'tenant_id');

Reference tables

Small lookup tables that every worker needs access to — pricing plans, country codes, feature flags. Replicate them with create_reference_table() and they can be joined with any distributed table without network round-trips.

-- Small lookup tables should be reference tables
CREATE TABLE plans (
  plan_id   serial PRIMARY KEY,
  name      text NOT NULL,
  max_seats int NOT NULL
);

-- Replicate to every worker node
SELECT create_reference_table('plans');

Co-located joins

When two distributed tables share the same distribution column, JOINs on that column execute entirely within each worker node. No data leaves the node. This is the key to Citus performance in multi-tenant workloads — and the reason your choice of distribution column matters so much.

-- Co-located join: both tables sharded by tenant_id
-- Executes entirely on each worker node, no cross-shard traffic
SELECT e.event_type, count(*)
FROM events e
JOIN user_actions ua ON ua.tenant_id = e.tenant_id
                    AND ua.event_id = e.event_id
WHERE e.tenant_id = 42
GROUP BY e.event_type;

Cross-shard aggregations

Queries that span all shards are parallelized across workers. Each worker computes a partial result, and the coordinator merges them. This is where the investment in multiple nodes earns its keep.

-- Aggregation across all tenants
-- Citus parallelizes across workers, merges on coordinator
SELECT date_trunc('hour', created_at) AS hour,
       count(*) AS event_count
FROM events
GROUP BY 1
ORDER BY 1 DESC
LIMIT 24;

Online shard rebalancing

When you add a worker node, rebalance shards without downtime. Citus moves shards using logical replication in the background.

-- After adding a new worker node, rebalance shards
SELECT citus_rebalance_start();

-- Monitor rebalance progress
SELECT * FROM citus_rebalance_status();

Cloud availability

Azure is currently the only major managed provider with native Citus support, which is not surprising given that Microsoft acquired Citus Data in 2019. For other cloud providers, Citus requires self-hosting on virtual machines or containers — a meaningful operational commitment that should factor into your decision.

How Gold Lapel relates

I should note: Gold Lapel sits between your application and PostgreSQL, seeing every query before it reaches the database. In a Citus deployment, that means we see queries before they arrive at the coordinator — which is precisely where intervention is most useful.

The relationship is complementary. Citus handles data distribution and parallel execution across nodes. Gold Lapel handles the quality of the queries themselves — materialized view recommendations, index suggestions, query rewriting. A poorly written query distributed across 10 nodes is still a poorly written query. It simply runs poorly in parallel. We would prefer to catch those patterns before Citus fans them out across your cluster.

For multi-tenant workloads, this is particularly valuable. Gold Lapel can identify patterns where queries are missing the distribution column in their WHERE clause — causing expensive cross-shard scatter that your monitoring may not immediately surface. It flags N+1 patterns generating unnecessary coordinator round-trips, and tracks per-tenant query latency to spot tenants whose workloads are quietly consuming disproportionate resources. The kind of thing one notices when paying close attention.