Sharding: shard keys and routing

conceptThe concept

Three ways to map a row to a shard:

• Hash: hash(key) % N. Spreads load evenly, but range queries on the key hit every shard, and changing N moves most rows unless you hash into many fixed logical shards and map those to servers.
• Range: key ranges per shard. Range scans stay local, but monotonically increasing keys (time, sequences) send all new writes to the last shard.
• Directory: a lookup table maps each key to a shard. Flexible (a big tenant can get its own server) but the directory must be fast and highly available.

A good shard key has high cardinality, spreads load evenly, appears in almost every query, and keeps rows that are used together on the same shard. In multi-tenant applications that is nearly always the tenant or customer id.

whyWhy it matters

A query that includes the shard key goes to one shard and scales linearly with the number of shards. A query without it must ask every shard and merge the results: its cost grows with the number of shards. The key decides which queries are which.

Use caseReal-world scenario

A B2B SaaS platform shards by tenant_id. Every table carries the tenant id, every query filters on it, and each tenant's orders, invoices and users live on one shard so joins and transactions stay local. The few global reports run on a separate analytics copy.

Try itExamples

Run and edit these in the interactive lesson.

SELECT (hashtext(customer_id::text) & 2147483647) % 4 AS shard,
       count(*) AS orders,
       count(DISTINCT customer_id) AS customers
FROM orders
GROUP BY 1
ORDER BY 1;

SELECT CASE WHEN customer_id <= 500 THEN 'shard_a'
            WHEN customer_id <= 1000 THEN 'shard_b'
            WHEN customer_id <= 1500 THEN 'shard_c'
            ELSE 'shard_d' END AS range_shard,
       count(*) AS orders
FROM orders
GROUP BY 1
ORDER BY 1;

BEGIN;
DO $$
BEGIN
  FOR i IN 0..3 LOOP
    EXECUTE format('CREATE SCHEMA shard_%s', i);
    EXECUTE format('CREATE TABLE shard_%s.orders (LIKE public.orders INCLUDING ALL)', i);
    EXECUTE format('INSERT INTO shard_%s.orders OVERRIDING SYSTEM VALUE
                    SELECT * FROM public.orders
                    WHERE (hashtext(customer_id::text) & 2147483647) %% 4 = %s', i, i);
  END LOOP;
END $$;

CREATE FUNCTION shard_for(customer integer) RETURNS text LANGUAGE sql IMMUTABLE AS $$
  SELECT 'shard_' || (hashtext(customer::text) & 2147483647) % 4
$$;

DO $$
DECLARE n integer;
BEGIN
  EXECUTE format('SELECT count(*) FROM %I.orders WHERE customer_id = 42', shard_for(42)) INTO n;
  RAISE NOTICE 'customer 42 has % orders on %', n, shard_for(42);
END $$;

SELECT 'shard_0' AS shard, count(*) FROM shard_0.orders
UNION ALL SELECT 'shard_1', count(*) FROM shard_1.orders
UNION ALL SELECT 'shard_2', count(*) FROM shard_2.orders
UNION ALL SELECT 'shard_3', count(*) FROM shard_3.orders;
ROLLBACK;

BEGIN;
CREATE TABLE shard_map (customer_id integer PRIMARY KEY, shard text NOT NULL);
INSERT INTO shard_map
SELECT id, CASE WHEN tier = 'platinum' THEN 'dedicated_1' ELSE 'shared_' || (id % 3) END
FROM customers;
SELECT shard, count(*) AS customers FROM shard_map GROUP BY shard ORDER BY shard;
ROLLBACK;

internalsUnder the hood

Systems that shard by hash usually hash into a fixed number of logical shards (Citus defaults to 32 per table) placed on fewer servers. Adding a server moves whole logical shards rather than rehashing every row, which is what makes rebalancing practical.

Avoid / preferDo this, not that

-- shard key: created_at (range by month)

-- shard key: customer_id (hash), partition by created_at inside each shard

-- shard key: order id

-- shard key: customer id, carried on orders, order_items and payments

PitfallsCommon mistakes

• Choosing a shard key that many queries do not include, turning common reads into fan-outs.
• Hashing directly onto physical servers, so adding a server rehashes most of the data.
• Forgetting global uniqueness: sequences per shard produce duplicate ids unless you use UUIDs, a shard prefix or an id service.

LimitsWhen not to use it

Do not shard data that fits comfortably on one well-tuned server with headroom for growth. Do not shard to isolate noisy tenants when a separate database per large tenant (a functional split) solves it with less machinery.

SecuritySecurity notes

With a directory or tenant-per-shard model, a routing bug sends a query to the wrong shard and can expose another tenant's data. Keep the tenant filter in every query even on a shard that should only contain one tenant, and consider row-level security as a second check.

ChallengeTry it yourself

WITH s AS (
  SELECT (hashtext(customer_id::text) & 2147483647) % 8 AS shard, sum(total) AS revenue
  FROM orders
  GROUP BY 1
)
SELECT count(*) AS shards, round(max(revenue) / min(revenue), 2) AS max_to_min
FROM s;

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