Understanding Citus: Distributed PostgreSQL Architecture and Core Concepts

Nodes

Citus

is a PostgreSQL extension that enables a cluster of database servers (nodes) to operate in a shared‑nothing architecture. Adding nodes increases storage capacity and CPU core count beyond a single machine.

Extension reference: https://www.postgresql.org/docs/current/external-extensions.html

Coordinator and Worker

Each cluster has a special coordinator node; all other nodes are worker nodes. Applications send queries to the coordinator, which forwards them to the appropriate workers and aggregates the results.

For each query the coordinator either routes it to a single worker or parallelises it across multiple workers, based on where the required data resides. The coordinator consults Citus‑specific metadata tables to obtain worker DNS names, health status, and data distribution.

Distributed Data

Table Types

The cluster supports three table types, each stored differently on nodes.

Type 1: Distributed Table

Distributed tables appear as ordinary tables to SQL statements but are horizontally partitioned across workers. Rows are stored in shard tables such as table_1001, table_1002, etc. The distribution column determines the shard a row belongs to; choosing it correctly impacts performance and functionality.

Type 2: Reference Table

Reference tables are distributed tables whose entire content resides in a single shard and is replicated on every worker. They have no distribution column. Because each worker holds a local copy, queries read reference data without network overhead. Writes, updates, and deletes on reference tables use two‑phase commit (2PC) to keep data consistent across workers.

2PC reference: https://en.wikipedia.org/wiki/Two-phase_commit_protocol

Type 3: Local Table

The coordinator node runs a regular PostgreSQL instance with the Citus extension installed. Ordinary (local) tables can be created on the coordinator without sharding, useful for small management tables such as user authentication data. Local tables coexist with distributed and reference tables; Citus stores cluster metadata in local tables.

Shards

Each shard is a smaller table on a worker that holds a subset of rows from a distributed table. The coordinator’s pg_dist_shard metadata table records the shard ID and its hash range ( shardminvalue, shardmaxvalue).

SELECT * FROM pg_dist_shard;
 logicalrelid | shardid | shardstorage | shardminvalue | shardmaxvalue
--------------+---------+--------------+---------------+---------------
 github_events| 102026  | t            | 268435456     | 402653183
 github_events| 102027  | t            | 402653184     | 536870911
 github_events| 102028  | t            | 536870912     | 671088639
 github_events| 102029  | t            | 671088640     | 805306367
 (4 rows)

To locate the shard containing a specific row, the coordinator hashes the distribution‑column value and selects the shard whose range includes that hash.

Shard Placement

Shard placement maps a shard ID to a specific worker’s table (e.g., github_events_102027). The mapping is stored in pg_dist_placement.

SELECT shardid, node.nodename, node.nodeport
FROM pg_dist_placement placement
JOIN pg_dist_node node ON placement.groupid = node.groupid
WHERE node.noderole = 'primary'::noderole
  AND shardid = 102027;

┌─────────┬───────────┬──────────┐
│ shardid │ nodename  │ nodeport │
├─────────┼───────────┼──────────┤
│  102027 │ localhost │     5433 │
└─────────┴───────────┴──────────┘

In the github_events example there are four shards; the number of shards per table is configurable.

Replication

Citus supports two replication modes: Citus‑based replication (creating backup shard placements) and PostgreSQL streaming replication (replicating the entire node database to a follower). Streaming replication is more efficient and transparent, requiring no involvement from Citus metadata tables.

Co‑location

Placing related shards on the same worker reduces network traffic for join queries, allowing them to execute within a single Citus node. For example, tables store, product, and purchase that share a store_id distribution column can be co‑located for efficient joins.

Parallelism

Distributing queries across multiple machines increases throughput and CPU core utilisation. Queries that touch shards spread evenly across workers can run at “real‑time” speed, especially when the final result is compact (e.g., aggregates).

Query Execution

When executing a multi‑shard query, Citus transforms the session into a set of per‑shard tasks, queues them, and runs each task when a connection to the relevant worker is available. The coordinator maintains a per‑session connection pool, limiting concurrent connections to a worker by the GUC citus.max_adaptive_executor_pool_size, configurable at the session level.

Short tasks benefit from sequential execution on the same connection, while long‑running tasks gain from parallelism. The GUC citus.executor_slow_start_interval controls the delay between connection attempts for tasks; setting it to 0 disables slow‑start behaviour.

After a task finishes, the session pool keeps the connection open for reuse, reducing the overhead of re‑establishing connections. The number of idle connections per worker is capped by citus.max_cached_conns_per_worker. Additionally, citus.max_shared_pool_size acts as a safety valve, limiting the total number of connections across all workers.

Example of shard‑placement query:

SELECT shardid, node.nodename, node.nodeport
FROM pg_dist_placement placement
JOIN pg_dist_node node ON placement.groupid = node.groupid
WHERE node.noderole = 'primary'::noderole
  AND shardid = 102027;

Result:

┌─────────┬───────────┬──────────┐
│ shardid │ nodename  │ nodeport │
├─────────┼───────────┼──────────┤
│  102027 │ localhost │     5433 │
└─────────┴───────────┴──────────┘

Configuration parameters relevant to query execution: citus.max_adaptive_executor_pool_size – maximum concurrent connections per worker for a session. citus.executor_slow_start_interval – delay (in milliseconds) between opening additional connections; 0 disables slow start. citus.max_cached_conns_per_worker – maximum idle connections retained per worker. citus.max_shared_pool_size – global limit on total connections across all workers.

These settings allow administrators to balance the benefits of parallelism against the overhead of database connections.