Act 7 in depth: sharding, co-location, and the seven-tier tradeoff cascade

Citus is deployed. Shard key is tenant_id. Most tenants are fast. Tenant Acme — 40% of all queries — saturates its shard while the other five shards sit idle. The cluster is effectively single-shard for Acme. The hot-shard failure was not caused by sharding; it was designed in on the day the shard key was chosen.

Act 7 — Three independent sharding decisions

Sharding at year-3 involves three decisions made independently:

Decision 1: Shard key. The shard key determines data distribution and query locality. For B2B multi-tenant SaaS, tenant_id is the natural choice: all queries for one tenant stay on one shard. For B2C, user_id. For time-series, a time bucket. The key must have high cardinality (so load spreads), uniform access patterns (so no single shard is hot), and be cheap to compute (so routing overhead is minimal).

Decision 2: Distribution method.

• Hash: rows are hashed to shards — even spread, range queries fan out across all shards.
• Range: rows in a key range go to a specific shard — range queries stay local, but a hot key range creates a hot shard.
• List: manual assignment — flexible, high operational cost.
• Directory: a lookup table maps key values to shards — most flexible, highest operational cost.

Decision 3: Co-location. Tables that join together must be placed on the same shard key so cross-shard joins stay local. In Citus: users, orgs, events, audit_log — all sharded by tenant_id — are co-located. A join between users and events for tenant 42 hits a single shard and returns at local latency. Without co-location, the same join fans out to all shards, collects results at the coordinator, and merges.

The federation tax

Once sharded, queries that span multiple shards must be federated: send the query to each shard, collect results, and merge on the application side or at the Citus coordinator.

A naive outer join across two shards becomes two inner queries (one per shard) and a client-side join. A GROUP BY that involves shards becomes a GROUP BY per shard, then a second GROUP BY on the aggregates. Each cross-shard query has latency = max(latency per shard) + merge time; tail latency dominates.

Citus handles federation with distributed query planning, but only for co-located tables (those with the same shard key). Reference tables (replicated to every worker node) are free — a small dimension table like countries or currencies can be replicated once and joined locally everywhere.

The hot-shard fix

When Acme accounts for 40% of all queries on a tenant_id-sharded cluster:

• Option A — logical split: Route Acme’s queries to a dedicated set of logical shards. The tenant-aware router multiplexes Acme across shard IDs that map to a dedicated physical worker. Other tenants share the remaining workers.
• Option B — physical cluster: Move Acme to its own Postgres cluster. Queries for Acme route there; the shared cluster handles all other tenants.

Both options require online resharding (citus_rebalance_table_shards), which uses logical replication for sub-second per-shard write pauses. Plan the window — catch-up replication adds load.

The seven-tier tradeoff cascade

Each act unlocks the next, but only if the previous is sound:

• Skip Act 1 (schema) → Acts 2–7 fight inefficient joins and mutable natural keys.
• Skip Act 2 (indexing) → Act 3 planning decisions are made blind; statistics are useless if the planner has no indexes to choose between.
• Skip Act 3 (stats) → Act 4 vacuum is the only lever; tables bloat while you guess at cardinality.
• Skip Act 4 (bloat) → Act 5 pool threads see table scans that are slower because dead tuples inflate heap pages.
• Skip Act 5 (pool sizing) → Act 6 migrations starve under connection storms; the connection handshake backlog prevents the migrate from acquiring locks in a timely window.
• Skip Act 6 (lock safety) → Act 7 sharding is impossible without downtime; every shard rebalance requires an exclusive lock if migrations were not designed for online operation.
• Skip Act 7 (when needed) → one tenant dominates the cluster and the others wait in queue.

The order is a constraint imposed by physics and Postgres internals, not a suggestion.