Setting Up Citus for Distributed PostgreSQL Queries

This guide walks through installing Citus and building a coordinator-plus-two-workers deployment that executes distributed queries as part of a broader federated query execution strategy within Cross-Partition Querying & Aggregation Strategies.

Unlike the postgres_fdw approach covered in executing federated queries across multiple PostgreSQL instances, Citus is not a bolt-on foreign-table layer: it extends the PostgreSQL planner itself, so the coordinator decomposes one SQL statement into per-shard fragments, runs them in parallel on the workers, and merges partial aggregates natively. The price is homogeneity — every node must run the same PostgreSQL major version with the same Citus version loaded.

Prerequisites

• Three hosts (or VMs) running the same PostgreSQL major version — 15 or 16 — with the Citus package repository configured on each
• Superuser access on all three instances to set shared_preload_libraries and create the extension
• Port 5432 open in both directions between the coordinator and every worker (pg_isready -h 10.0.1.21 -p 5432)
• pg_hba.conf on each worker allowing scram-sha-256 connections from the coordinator’s IP, and .pgpass or certificate auth for coordinator-to-worker traffic
• A distribution-column candidate identified for each large table — typically tenant_id or another high-cardinality key present in most WHERE clauses
• Enough free disk on workers for shard copies during rebalancing (peak usage roughly doubles per moving shard)
• A maintenance window for the initial create_distributed_table call on existing tables, which copies data into shards

Step 1 — Install Citus and preload the extension library

Citus hooks into the PostgreSQL planner, so its shared library must be loaded at server start on every node — coordinator and workers alike — before the extension can be created. Install the package, set shared_preload_libraries, restart, and create the extension in the target database.

# Run on the coordinator AND both workers (Ubuntu, Citus repo configured)
sudo apt-get install -y postgresql-16-citus-13.0

sudo -u postgres psql -c "ALTER SYSTEM SET shared_preload_libraries = 'citus';"
sudo systemctl restart postgresql@16-main

sudo -u postgres createdb orders_db
sudo -u postgres psql -d orders_db -c "CREATE EXTENSION citus;"
sudo -u postgres psql -d orders_db -c "SELECT citus_version();"

Operational note: If shared_preload_libraries already lists other libraries (for example pg_stat_statements), append rather than replace: ALTER SYSTEM SET shared_preload_libraries = 'citus,pg_stat_statements'. Citus must appear in the list before the restart or CREATE EXTENSION citus fails with citus.so not loaded.

DBA tip: Pin the exact Citus package version in your configuration management. All nodes must run the same Citus minor version; a mixed-version deployment fails at metadata sync with obscure unsupported protocol errors rather than a clear version message.

Step 2 — Register the coordinator and add two workers

The coordinator learns about workers through metadata functions, not configuration files. First tell the deployment its own coordinator address (required so workers can reach back for metadata sync), then register each worker with citus_add_node.

-- Run on the coordinator, inside orders_db
SELECT citus_set_coordinator_host('10.0.1.10', 5432);

SELECT citus_add_node('10.0.1.21', 5432);
SELECT citus_add_node('10.0.1.22', 5432);

-- Confirm both workers are registered and active
SELECT * FROM citus_get_active_worker_nodes();

Operational note: citus_add_node validates connectivity immediately — it opens a connection from the coordinator to the worker as the postgres superuser (or the current user). If it hangs, the problem is almost always pg_hba.conf on the worker or a firewall rule, not Citus itself.

SRE tip: Register workers by DNS name or a stable virtual IP rather than a raw instance IP where possible. Replacing a failed worker then becomes citus_update_node plus a DNS flip instead of a full drain-and-rebalance cycle.

Step 3 — Distribute a table with a shard key

create_distributed_table splits a table into citus.shard_count hash shards on the chosen distribution column and spreads them across the registered workers. Set the shard count before distributing — it is fixed for the table’s colocation group afterwards.

-- Shard count is read at distribution time; set it first
ALTER SYSTEM SET citus.shard_count = 32;
SELECT pg_reload_conf();

CREATE TABLE orders (
  order_id    BIGINT GENERATED ALWAYS AS IDENTITY,
  tenant_id   INT          NOT NULL,
  total       NUMERIC(12,2) NOT NULL,
  currency    CHAR(3)      NOT NULL,
  created_at  TIMESTAMPTZ  NOT NULL DEFAULT now(),
  PRIMARY KEY (tenant_id, order_id)   -- must include the distribution column
);

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

Operational note: Every unique index and primary key on a distributed table must include the distribution column, because uniqueness is only enforceable within a single shard. That is why the primary key above is (tenant_id, order_id) rather than order_id alone.

DBA tip: Choose a shard count that is a comfortable multiple of your target worker count — 32 shards over 2 workers gives 16 per node and leaves headroom to grow to 4, 8, or 16 workers by rebalancing alone. Too few shards caps future parallelism; a few hundred is the practical ceiling before per-shard overhead bites.

Step 4 — Colocate related tables and create reference tables

Joins between distributed tables are only pushed down to workers when both tables share the same distribution column and colocation group — otherwise Citus must repartition data over the network at query time. Distribute child tables with colocate_with, and turn small lookup tables into reference tables that are replicated in full to every worker.

CREATE TABLE order_items (
  item_id    BIGINT GENERATED ALWAYS AS IDENTITY,
  tenant_id  INT           NOT NULL,
  order_id   BIGINT        NOT NULL,
  sku        TEXT          NOT NULL,
  price      NUMERIC(12,2) NOT NULL,
  PRIMARY KEY (tenant_id, item_id)
);

-- Same distribution column, same colocation group as orders
SELECT create_distributed_table('order_items', 'tenant_id',
                                colocate_with => 'orders');

-- Small lookup table: replicate to all workers instead of sharding
CREATE TABLE currencies (
  code CHAR(3) PRIMARY KEY,
  usd_rate NUMERIC(10,6) NOT NULL
);
SELECT create_reference_table('currencies');

Operational note: Foreign keys between distributed tables are only allowed inside a colocation group and must include the distribution column on both sides. A foreign key from any distributed table to a reference table, however, is always valid — that is the main reason currencies is a reference table rather than a plain local table on the coordinator.

SRE tip: Reference tables are written through two-phase commit to every worker, so keep them small and low-churn. A “lookup” table receiving thousands of updates per second will serialise those writes across the whole deployment.

Step 5 — Run a cross-shard aggregation and read the distributed EXPLAIN

A GROUP BY over the whole table fans out to all 32 shards; each worker computes partial aggregates locally and the coordinator merges them. The distributed EXPLAIN shows exactly how the query decomposes — check the task count and the per-shard fragment before trusting performance numbers.

EXPLAIN (VERBOSE, COSTS OFF)
SELECT o.currency,
       count(*)                    AS order_count,
       sum(o.total * c.usd_rate)   AS revenue_usd
FROM orders o
JOIN currencies c ON c.code = o.currency
WHERE o.created_at >= now() - interval '30 days'
GROUP BY o.currency;

The plan should open with a Custom Scan (Citus Adaptive) node reporting Task Count: 32, and each task’s fragment must contain the WHERE filter and a partial GROUP BY:

Custom Scan (Citus Adaptive)
  Output: remote_scan.currency, ...
  Task Count: 32
  Tasks Shown: One of 32
  ->  Task
        Query: SELECT o.currency, count(*) ... FROM orders_102008 o
               JOIN currencies_102040 c ON ...
               WHERE o.created_at >= ... GROUP BY o.currency

Operational note: A query that filters on the distribution column (WHERE tenant_id = 42) should show Task Count: 1 — single-shard routing. If it shows 32, the filter is not being recognised (commonly a type mismatch, e.g. tenant_id = '42'::text), and every query is paying full fan-out cost.

SRE tip: Watch citus.max_adaptive_executor_pool_size (default 16) — it caps per-query connections from coordinator to each worker. Under high concurrency, total worker connections approximate active queries x pool size, so budget worker max_connections accordingly or front the workers with a pooler.

Step 6 — Add a node and rebalance shards

Scaling out is a two-step operation: register the new worker, then start the rebalancer. The rebalancer moves shards using logical replication so the deployment stays online; progress is observable from a monitoring view.

-- Register the new worker
SELECT citus_add_node('10.0.1.23', 5432);

-- Move shards onto it in the background (non-blocking)
SELECT citus_rebalance_start();

-- Watch progress
SELECT * FROM citus_rebalance_status();

Operational note: citus_rebalance_start() returns immediately and runs in the background; citus_rebalance_status() reports state (running, finished), the current move, and bytes copied. Cancel a misbehaving rebalance with citus_rebalance_stop() — completed moves stay in place, in-flight ones roll back.

DBA tip: Each shard move creates a logical replication slot on the source worker. If a move stalls (long-running transaction on the source blocking catch-up), WAL accumulates behind the slot and can fill the disk. Alert on pg_replication_slots.restart_lsn age on workers during any rebalance window.

Verification

Confirm the topology, shard placement, and data distribution before routing production traffic at the coordinator:

-- 1. Distributed tables and their colocation groups
SELECT table_name, citus_table_type, distribution_column, shard_count
FROM citus_tables;

-- 2. Shard placement per worker — counts should be balanced
SELECT nodename, count(*) AS shard_count,
       pg_size_pretty(sum(shard_size)) AS total_size
FROM citus_shards
GROUP BY nodename
ORDER BY nodename;

-- 3. End-to-end smoke test: insert routes to one shard, aggregate fans out
INSERT INTO orders (tenant_id, total, currency) VALUES (42, 99.50, 'EUR');
SELECT count(*) FROM orders;

Expected output for check 1: orders and order_items listed as distributed with distribution_column = tenant_id and shard_count = 32; currencies listed as reference. Expected output for check 2 with three workers: roughly 10–11 shards per node with sizes within a few percent of each other. Check 3 must return the inserted row in the count with no error — if the INSERT fails with a constraint error, a unique index without the distribution column slipped through Step 3.

Failure mode table

FAQ

Related

• Federated Query Execution — parent guide to coordinator architecture, pushdown, and result merging across distributed nodes
• Running Federated Queries Across Shards with Trino — federate existing independent shards without migrating them into a Citus deployment
• Querying Sharded PostgreSQL with DuckDB — lightweight ad-hoc analytics over the same shards when a coordinator service is overkill