Co-location and Citus: the invariant that makes sharding usable

A team migrates to Citus and is thrilled that queries for individual tenants are as fast as before. Then they add a new table without the tenant_id column, and a new join query takes 500ms instead of 5ms. The schema review catches why: one table was not co-located.

Citus architecture

Citus is a Postgres extension that turns a cluster of Postgres instances into one logical sharded database:

• Coordinator: one node that holds the cluster metadata (shard map, table distribution info), parses queries, plans the distributed execution, and routes results back to the client.
• Workers: N nodes that hold the actual data (shards). Workers execute Postgres queries locally and return results to the coordinator.
• Client: connects to the coordinator as if it were a normal Postgres. No driver changes required.

Client → Coordinator (holds metadata, plans, routes)
              ↓               ↓               ↓
          Worker 0        Worker 1        Worker 2
        (shards 0-10)  (shards 11-21)  (shards 22-31)

Three table types

-- Distribute tenant-scoped tables by tenant_id
SELECT create_distributed_table('users',       'tenant_id');
SELECT create_distributed_table('projects',    'tenant_id', colocate_with => 'users');
SELECT create_distributed_table('tasks',       'tenant_id', colocate_with => 'users');
SELECT create_distributed_table('comments',    'tenant_id', colocate_with => 'users');
SELECT create_distributed_table('attachments', 'tenant_id', colocate_with => 'users');

-- Replicate small lookup tables to every worker
SELECT create_reference_table('plans');
SELECT create_reference_table('feature_flags');
SELECT create_reference_table('countries');

-- tenants table stays local on coordinator (control plane)

Co-location: the central invariant

Co-location means that tables distributed by the same key have their corresponding shards on the same physical worker. If orders and payments are both distributed by tenant_id, then all of tenant 42’s orders and all of tenant 42’s payments live on Worker 1.

A query joining orders and payments filtered by tenant_id = 42:

• With co-location: the planner pushes the join to Worker 1, which executes it like a normal Postgres join — one machine, full index use, single-digit milliseconds.
• Without co-location (e.g., payments distributed by payment_id instead): the coordinator must gather partial results from all workers and merge them. P99 latency = slowest worker. Every worker participates. 10–100× slower.

Co-location is not a performance optimization — it is the design contract that makes a sharded system behave like a single Postgres for the 99% case.

Cross-shard queries and their mitigation

Queries without a shard key filter fan out to all shards. Examples:

• “List all users with email ending in @enterprise.com” (no tenant_id)
• “Count total orders today across all tenants” (cross-tenant analytics)
• “Find a user by their email for login” (email, not tenant_id)

For OLTP:

1. Redesign the API: almost all OLTP queries should carry tenant_id. Login lookups by email need a separate index or a small lookup service that resolves email → tenant_id first.
2. CDC to OLAP: cross-tenant analytics should run on a separate analytical store (ClickHouse, BigQuery) fed by Change Data Capture. Never run global aggregates on the OLTP cluster.
3. Rate-limit fan-out endpoints: for the rare legitimate cross-shard operation, rate-limit and document its cost.

The senior metric: cross-shard queries should be < 1–2% of OLTP traffic. Above that, the schema has drifted from co-location and needs review.