Distributed PostgreSQL: Citus and co-location

conceptThe concept

In Citus (an open-source PostgreSQL extension) a coordinator node holds metadata and plans queries; worker nodes hold shards, which are ordinary PostgreSQL tables.

• Distributed tables are hashed on a distribution column into shards.
• Reference tables (small lookup tables such as products or countries) are copied to every node, so joins to them are always local.
• Co-located tables share the distribution column and shard boundaries, so orders and order_items for customer 42 live on the same node and join locally.

A query with the distribution column in its WHERE clause is routed to one shard. Other queries are split into fragments that run on every shard in parallel and are merged on the coordinator.

Other paths: sharding in the application with a router library, managed sharded PostgreSQL services from cloud providers, and distributed SQL databases that speak the PostgreSQL protocol (YugabyteDB, CockroachDB). The last group is wire-compatible but is not PostgreSQL underneath, so extensions and some behaviours differ.

whyWhy it matters

Citus keeps the PostgreSQL you already know (types, extensions, SQL, tools) while removing the single-server limit. The price is that your schema must be designed around the distribution column.

Use caseReal-world scenario

A multi-tenant analytics product with 4,000 tenants outgrows one server. It distributes every tenant-scoped table by tenant_id, makes plans and features reference tables, and keeps per-tenant dashboards as routed single-shard queries.

Try itExamples

Run and edit these in the interactive lesson.

SELECT c.relname,
       bool_or(a.attname = 'customer_id') AS has_customer_id
FROM pg_class c
JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum > 0 AND NOT a.attisdropped
WHERE c.relnamespace = 'public'::regnamespace AND c.relkind = 'r'
GROUP BY c.relname
ORDER BY 2, 1;

BEGIN;
CREATE TABLE d_orders AS SELECT * FROM orders;
ALTER TABLE d_orders ADD PRIMARY KEY (customer_id, id);
CREATE TABLE d_order_items AS
SELECT o.customer_id, i.*
FROM order_items i JOIN orders o ON o.id = i.order_id;
ALTER TABLE d_order_items
  ADD PRIMARY KEY (customer_id, order_id, line_no),
  ADD FOREIGN KEY (customer_id, order_id) REFERENCES d_orders (customer_id, id);
SELECT count(*) AS items_for_customer_42 FROM d_order_items WHERE customer_id = 42;
ROLLBACK;

CREATE EXTENSION citus;
SELECT citus_add_node('worker-1', 5432);
SELECT citus_add_node('worker-2', 5432);

SELECT create_distributed_table('orders', 'customer_id');
SELECT create_distributed_table('order_items', 'customer_id', colocate_with => 'orders');
SELECT create_reference_table('products');

-- routed: one shard on one worker
SELECT * FROM orders WHERE customer_id = 42;

-- fanned out to every shard, merged on the coordinator
SELECT status, count(*) FROM orders GROUP BY status;

-- move shards onto a newly added worker
SELECT citus_rebalance_start();

internalsUnder the hood

Foreign keys and unique constraints between distributed tables are only enforceable when they include the distribution column, because each shard can only check its own rows. That is why primary keys become composite, such as (customer_id, id), and why the tenant id is denormalized onto child tables.

Avoid / preferDo this, not that

SELECT create_distributed_table('orders', 'customer_id');
SELECT create_distributed_table('order_items', 'order_id');

SELECT create_distributed_table('orders', 'customer_id');
SELECT create_distributed_table('order_items', 'customer_id', colocate_with => 'orders');

SELECT create_distributed_table('countries', 'code');

SELECT create_reference_table('countries');

PitfallsCommon mistakes

• Distributing tables before adding the tenant id to child tables, then discovering joins are not co-located.
• Running many cross-shard transactions and expecting single-node latency.
• Treating a wire-compatible distributed SQL database as PostgreSQL and finding missing extensions or different isolation behaviour in production.

LimitsWhen not to use it

Skip distributed PostgreSQL for workloads dominated by cross-entity queries with no natural tenant key, or for data that fits on one server. A read replica or a columnar analytics copy is usually the better answer for heavy reporting.

SecuritySecurity notes

The coordinator connects to workers with its own credentials; protect those connections with TLS and restrict pg_hba.conf on workers to the coordinator and other workers. Row-level security policies must be created consistently on every node.

ChallengeTry it yourself

SELECT k.conrelid::regclass AS child_table, k.conname
FROM pg_constraint k
WHERE k.contype = 'f'
  AND k.confrelid = 'orders'::regclass
  AND NOT EXISTS (
    SELECT 1 FROM pg_attribute a
    WHERE a.attrelid = k.conrelid AND a.attname = 'customer_id' AND NOT a.attisdropped
  );

InterviewQuestions you should be able to answer

QuizCheck your understanding

4 questions with instant feedback in the interactive lesson; a wrong answer goes into your review deck.

Go deeperOfficial documentation

• postgres_fdw
• PREPARE TRANSACTION
• Citus documentation