Partitioning & Sharding

The 30-second version

The words get used loosely. Partitioning is the general idea of splitting data; sharding usually means the pieces live on different servers. A DatabaseDatabaseA program whose whole job is storing data safely and finding it again quickly — Postgres, MySQL, MongoDB, Redis. Your app usually talks to it over a network connection; embedded databases like SQLite run inside your program instead. can also be partitioned on one machine (into several files), which §10 shows in PostgreSQLPostgreSQL (Postgres)A popular open-source relational SQL database, known for correctness and features. Used throughout this site's database examples..

Splitting is not copying

People often mix up partitioning with replication, which means keeping full copies of the same data on several machines (a ReplicaReplicaA full copy of a database kept on another machine and updated as the original changes, for reading load and for taking over if the original dies. is one such copy; the replication topic covers it). They solve different problems, and real systems use both. Flip between the three setups:

What the rest of this topic covers

1. Why a single database runs out of room (§02).
2. How to split: by range, by hash, or with a lookup table (§03), and how to pick the key (§04).
3. How to find the right shard for a request (§05).
4. What goes wrong: one shard on fire (§06), queries that have to ask every shard (§07).
5. How to grow: moving data to new machines without downtime (§08, §09).
6. Real tools and staff-level advice, including when not to shard (§10, §11).

Why one database stops being enough

Wall 1: the data doesn't fit

Say your app stores 2 KB per order and takes 5 million orders a day. That's about 10 GB a day, or 3.6 TB a year, before indexesIndex (database)An extra, sorted data structure the database keeps next to a table so it can find rows by a column without reading the whole table — like the index at the back of a book.. A big server can hold tens of terabytes, so this wall is further away than people think, but it's real: and long before the disk is full, backups, restores and adding a column take hours.

Wall 2: the busy data doesn't fit in memory

A database is fast when the rows people are asking for are already in RAM. Reading from an SSD vs HDDSSD vs HDDSolid-state drives read a random block in ~0.1 ms; spinning hard disks need ~5–10 ms because a head must physically move. Both are far slower than RAM (~0.0001 ms). is roughly 100 times slower than reading memory (approximate; it depends on the hardware). The rows that are in active use are called the working set. When it grows past the server's RAM, more and more reads go to disk and LatencyLatencyHow long one request or job takes from the moment it arrives until it's finished — including any time spent waiting in a queue. climbs.

Wall 3: too many writes

Reads are easy to spread: add replicasReplicaA full copy of a database kept on another machine and updated as the original changes, for reading load and for taking over if the original dies. and send reads to them. Writes are not, because every replica has to apply every write too. In the usual setup one main server, the primary, accepts all the writes and the replicas copy them. If that primary can apply 20,000 writes a second (a rough number for a well-tuned box) and you need 60,000, more copies won't help. You need three machines that each take a different third of the writes. That's partitioning.

What to try before sharding

Sharding is powerful and expensive: every query, migration and backup gets harder. So the usual order is:

Scaling up has a ceiling (the biggest machine you can buy) and a price curve that gets steep near the top. Sharding is the opposite trade, called scaling outScaling out (horizontal scaling)Handling more traffic by adding more machines that each run a copy of your app, instead of buying one bigger machine (scaling up). For databases, each machine holds part of the data.: many ordinary machines, no ceiling in principle, but much more complexity.

Three ways to split: range, hash, directory

Range: like the phone book

With Range partitioningRange partitioningSplitting data by ranges of the key: A–F on one shard, G–M on the next, or one month per shard. Neighbouring keys stay together, so “everything between X and Y” is cheap., each shard owns a continuous slice of key values: names A–F, G–M…, or one month of orders per shard.

• Good: neighbouring keys sit together. A range queryRange queryA query asking for all values between two bounds — created_at between March 1 and March 2, price < 100 — instead of one exact value. such as "all orders from last week" goes to one or two shards.
• Bad: if new keys always grow, every new row lands in the last range. Order IDs that count up, and timestamps, are like that (a monotonic keyMonotonic keyA key that only ever goes up: auto-increment IDs, timestamps. With range partitioning, every new row lands in the last range.). The newest shard takes all the writes while the others sit idle: a hot spotHot key / hot spotA single key (or a narrow range) that gets far more traffic than the rest, like a celebrity's profile. Because one key always lives on one shard, that shard overloads while the others idle..

Hash: scramble first, then split

With Hash partitioningHash partitioningChoosing the shard from a hash of the key, e.g. hash(user_id) % 1024. Keys spread evenly, but neighbouring keys end up on different shards., you first run the key through a hash functionHash (for sharding)A function that turns a key like "tenant:42" into a big, evenly spread number. The same key always gives the same number; similar keys give very different numbers., which turns it into a big, random-looking number. Then you pick a shard from that number, for example hash(key) % 4 (the remainder after dividing by 4, which is always 0–3).

• Good: consecutive keys scatter, so writes spread evenly even when IDs count up.
• Bad: the same scattering destroys order. "Orders 300–340" are now on every shard, so a range query has to ask all of them.

(Plain % 4 has a nasty problem when you go from 4 shards to 5; §08 covers it. For now, read it as "a hash picks the shard.")

Directory: a lookup table

With Directory (lookup) partitioningDirectory (lookup) partitioningKeeping an explicit table that says which shard each key (or group of keys) lives on. Flexible, since you can move any key by changing one row, but the table must be stored and looked up., you keep an explicit table: customer acme → shard 0, globex → shard 1, and so on.

• Good: total control. Move one customer by changing one row; give a huge customer its own shard.
• Bad: every request needs a lookup first, so the table must be fast, cached, and never lost.

Try all three on the same data

Each order has an ID (counting up) and a customer. Press ▶ to add orders and watch the "last 40 writes" bars. Then run the two queries under each scheme.

What you should have seen:

Notice that the widget's directory splits by customer, not by order ID. That, not the directory itself, is why the globex query is cheap: hashing by customer would make it cheap too. The scheme (range, hash, directory) and the key (order ID, customer, …) are two separate choices. §04 is about the key.

Choosing a shard key

Three questions to ask about a candidate key

A worked example: TaskBoard

TaskBoard is a project-management app sold to companies. Each customer company is a tenant, which makes it a multi-tenantMulti-tenant (tenant)One app serving many customer organisations (tenants) from shared infrastructure, like Slack workspaces or Shopify stores. Each tenant only sees its own data. app. It has three main tables: projects, tasks and users, and every row belongs to one tenant.

The two most common requests look like this:

-- 1. Show a tenant's board: their open tasks, with project names
SELECT t.title, p.name
FROM tasks t
JOIN projects p ON p.id = t.project_id
WHERE t.tenant_id = 42 AND t.status = 'open';

-- 2. Open one task (the URL is /acme/tasks/123, so we know the tenant)
SELECT * FROM tasks WHERE tenant_id = 42 AND id = 123;

Both queries contain tenant_id. That's the clue. Now compare the candidates. The data is realistic: 40 tenants, and the biggest one, acme, is about 23% of everything.

Why tenant_id wins (with one fix)

Sharding by tenant_id puts all of a tenant's projects, tasks and users on the same shard. This is called co-locationCo-locationPutting rows that are used together (a tenant's projects, tasks and users) on the same shard by giving them the same shard key, so JOINs and transactions stay on one machine., and it buys three things at once:

1. The JOIN in query 1 runs inside one database, exactly as it did before sharding.
2. A Database transactionDatabase transactionA group of database changes that happen all together or not at all (COMMIT keeps them, ROLLBACK undoes them). How much other users can see or interfere with the half-done state depends on the isolation level. that touches several tables for one tenant (create a task and update the project's counter) stays on one machine, so it keeps its all-or-nothing guarantee. Cross-shard transactions are hard (§07).
3. The app barely changes. Every request already knows its tenant, from the URL or the logged-in user.

The weakness is the one the widget showed: one whale tenantWhale tenantA customer so much bigger than the rest that it can overload whichever shard it lands on. Often given its own dedicated shard. makes its shard much fuller than the rest. The usual fix is a hybrid: hash small tenants, and add directory entries that pin each whale to a dedicated shard. If a single whale outgrows even a whole machine, split it by a second key (such as project_id) as a special case.

Other common choices

Finding the right shard: routing

Three places to put the lookup

In every version, the routing table itself lives in a small metadata storeMetadata storeA small, highly reliable database (often etcd or ZooKeeper) that holds the routing table and tells routers when it changes. (etcd, ZooKeeper, or a special table), and everyone who routes keeps a copy. When data moves, the table changes and the copies must update. A router with an old copy sends queries to the wrong place, so real systems also let a shard answer "not mine" instead of returning wrong results:

Build one: a client-side router

Here's the smallest useful version: the app hashes the shard key to one of 1,024 fixed partitions, then a table says which database server owns each partition. Why not hash straight to a server? Because then adding a server would reshuffle almost every key; §08 explains in detail. With this design, moving data means changing table rows, never the formula.

package shard

import (
	"database/sql"
	"fmt"
	"hash/fnv"
)

// NumPartitions is fixed forever. Nodes come and go; partitions don't.
const NumPartitions = 1024

// PartitionOf maps a shard key to a partition number 0…1023.
func PartitionOf(key string) int {
	h := fnv.New32a()    // FNV-1a: fast, and the same result in every process
	h.Write([]byte(key)) // a hash's Write never returns an error
	return int(h.Sum32() % NumPartitions)
}

// Router knows which node owns each partition, and holds one pool per node.
type Router struct {
	owner [NumPartitions]string // partition → node name, e.g. "db-a"
	pools map[string]*sql.DB    // node name → connection pool
}

func NewRouter(owner [NumPartitions]string, pools map[string]*sql.DB) (*Router, error) {
	for p, node := range owner {
		if _, ok := pools[node]; !ok { // ← catch typos at startup, not at 3 a.m.
			return nil, fmt.Errorf("partition %d: unknown node %q", p, node)
		}
	}
	return &Router{owner: owner, pools: pools}, nil
}

// Node answers "which machine holds this key?"
func (r *Router) Node(key string) string { return r.owner[PartitionOf(key)] }

// DB returns the pool to run this key's queries on.
func (r *Router) DB(key string) *sql.DB { return r.pools[r.Node(key)] }

Using it looks like ordinary database code, with one extra step to pick the pool. The tenant ID in the query is the same value you routed by:

key := fmt.Sprintf("tenant:%d", tenantID)
var name string
err := router.DB(key).QueryRowContext(ctx,
	"SELECT name FROM projects WHERE tenant_id = $1 AND id = $2", tenantID, projectID,
).Scan(&name)

$1, $2 are placeholders: the driver sends the values separately from the SQL text, which also protects against SQL injectionSQL injectionAn attack where user input pasted into a SQL string changes the query itself (e.g. a name like ' OR 1=1 --). Prevent it by sending values separately as placeholders (?)..

Test it

Routing bugs are silent: a wrong shard just returns "not found". So test the properties that matter. The full test files have a few more cases; these are the two that teach the most:

// Golden values: the JS router must produce exactly the same numbers.
func TestPartitionGolden(t *testing.T) {
	for key, want := range map[string]int{"tenant:1": 564, "tenant:42": 491, "acme": 431} {
		if got := PartitionOf(key); got != want {
			t.Errorf("PartitionOf(%q) = %d, want %d", key, got, want)
		}
	}
}

func TestMovingOnePartitionMovesOnlyItsKeys(t *testing.T) {
	owners := evenOwners("db-a", "db-b", "db-c") // helper: round-robin table
	pools := fakePools("db-a", "db-b", "db-c", "db-d") // helper: nil pools
	before, err := NewRouter(owners, pools)
	if err != nil {
		t.Fatal(err)
	}
	owners[7] = "db-d" // ← hand partition 7 to a new node (arrays copy, so before is unchanged)
	after, err := NewRouter(owners, pools)
	if err != nil {
		t.Fatal(err)
	}
	for i := range 30_000 {
		k := fmt.Sprintf("tenant:%d", i)
		if before.Node(k) != after.Node(k) && PartitionOf(k) != 7 {
			t.Fatalf("%s moved but is in partition %d", k, PartitionOf(k))
		}
	}
}

We ran both suites (Go 1.24, Node 22): both pass, and both print the same partition numbers. Spreading 30,000 tenant keys over three nodes gave 9,995 / 10,145 / 9,860, and handing partition 7 to a new node moved 45 keys, all from partition 7. (A perfect hash would move 30,000 ÷ 1,024 ≈ 29; real hashes are a bit lumpy.)

Hot keys: one shard on fire

The celebrity problem

A social app shards posts by post ID. A pop star with 100 million followers posts a photo. Within a minute, hundreds of thousands of people open it and tap "like". Every one of those requests contains the same key, post:42, so every one of them goes to the same shard.

Adding more shards does nothing here. The key still lives on exactly one of them. This is why it's also called skew: the load is lopsided, no matter how many machines you have.

Try it

Move the slider to make the post go viral. The salting toggle uses 8 sub-keys (2 per shard), the same number as the code further down. Then try the two fixes, once for reads (people viewing) and once for writes (people liking).

What the simulator shows:

Fix 1 for hot reads: cache it

The same post being read a million times gives a million identical answers. Put a CacheCacheA small, fast store of answers you already computed or fetched, so the next request for the same thing skips the slow work. The hard part is knowing when a cached answer is out of date. in front, either an in-memory one in each app server or a shared one like RedisRedisA popular in-memory data server: a fast key–value store and cache that also offers counters, sorted sets, pub/sub channels and streams (an append-only log for messages).. Even a cache that keeps the post for one second turns 100,000 reads per second into one database read per second per app server.

Fix 2 for hot writes: salt the key

Key saltingKey salting (key splitting)Spreading one hot key over N sub-keys (post:42#0 … post:42#7) that land on different shards. Writes pick one sub-key; reads must combine all N. splits one logical key into N physical sub-keys: post:42#0, post:42#1, …, post:42#7. Each has a different hash, so they land on different shards (we checked: with 4 nodes and our router from §05, the 8 sub-keys of post:42 spread over all 4). Each write picks one sub-key at random; a read of the total adds all of them up.

const saltN = 8 // how many sub-keys one hot counter is split into

// AddLike increments ONE randomly chosen sub-key: writes spread over 8 places.
func AddLike(ctx context.Context, postID int64) error {
	key := fmt.Sprintf("post:%d#%d", postID, rand.IntN(saltN)) // "post:42#5"
	_, err := router.DB(key).ExecContext(ctx, `
		INSERT INTO like_counts (key, n) VALUES ($1, 1)
		ON CONFLICT (key) DO UPDATE SET n = like_counts.n + 1`, key)
	return err
}

// LikeCount must read ALL sub-keys and add them up: reads fan in from 8 places.
func LikeCount(ctx context.Context, postID int64) (int64, error) {
	var total int64
	for i := range saltN {
		key := fmt.Sprintf("post:%d#%d", postID, i)
		var n int64
		err := router.DB(key).QueryRowContext(ctx,
			"SELECT n FROM like_counts WHERE key = $1", key).Scan(&n)
		if err != nil && !errors.Is(err, sql.ErrNoRows) { // a missing sub-key just means 0
			return 0, err
		}
		total += n
	}
	return total, nil
}

The Go version reads the sub-keys one after another for clarity; in production you'd run them concurrently like the JS one. Either way it's 8 lookups per read, which is why §07's tail-latency math applies here too.

Other tools against hot spots

• Batch the writes. Collect likes in memory for 1 second and write n = n + 437 once. Fewer, bigger writes.
• Split the range. If the hot spot is a range (today's orders), move that range to its own shard, or split it further (§08).
• Rate-limit or shed. If one tenant is hammering a shard, cap that tenant so it can't take everyone else down with it (load shedding).

Queries and transactions across shards

Scatter-gather

"Top 10 most expensive orders this week, across all tenants" doesn't name a tenant, so the router can't pick a shard. It runs Scatter-gatherScatter-gatherRunning a query by sending it to every shard (scatter), waiting for all the answers, and merging them (gather). Needed when the query doesn't include the shard key.:

1. Scatter: send the query to all N shards in parallel.
2. Each shard computes its own top 10.
3. Gather: wait for all N answers, merge them, keep the overall top 10.

That works, and for an occasional admin query it's fine. The trouble is speed: the query is only as fast as the slowest shard. Each shard is usually quick, but every machine occasionally hiccups (a garbage-collectionGarbage collector (GC)Part of the language runtime that automatically frees memory nobody can reach anymore. Anything still referenced — like a stuck goroutine — is never freed. pause, a slow disk, another program on the same hardware hogging it). With one shard you rarely notice. With 50, someone is almost always hiccuping.

The maths is short. If each shard is slow 1% of the time, a query is fast only if every shard is fast: 0.99 × 0.99 × … = 0.99^N. For N = 50 that's 0.61, so 39% of queries are slow. A rare problem per shard becomes a common problem per query. This is Tail latencyTail latencyThe latency of the slowest few requests — usually reported as p99 or p99.9. It's what your unluckiest users feel, and it grows much faster than the average under load., and it's why you measure p99p99 (99th percentile)The value that 99% of measurements are below — i.e. how bad the slowest 1% is. Averages hide slow outliers; p99 shows them. and not the average (more in Worker Pools §05).

Secondary indexes: local or global?

Users are sharded by user ID. Logging in looks a user up by email. That needs a secondary indexSecondary indexAny index other than the one the table is stored in. Its entries point back to the row: by row address in Postgres, by primary key value in InnoDB. on email, and a sharded database has two ways to build it:

Here is what changing bob@'s email to zed@ looks like in each kind of global index. The user row is on shard 0; the old index entry (b) is on shard 0 and the new one (z) is on shard 2:

A common hand-built version of a global index is a separate lookup table, email → user_id, itself sharded by email. Login reads it first, then goes to the right user shard.

Transactions across shards

Moving $100 from Alice to Bob is one Database transactionDatabase transactionA group of database changes that happen all together or not at all (COMMIT keeps them, ROLLBACK undoes them). How much other users can see or interfere with the half-done state depends on the isolation level. on a single database: both changes happen, or neither does (the "A" in ACIDACIDThe four guarantees of a database transaction: Atomic (all or nothing), Consistent (rules hold), Isolated (concurrent transactions don't see each other's half-done work), Durable (once committed, it survives a crash).). Now put Alice on shard A and Bob on shard B. They're two separate databases, and each commits on its own:

Three ways out, from best to worst:

1. Don't need one. Choose the shard key so things that change together live together (§04). Most SaaS transactions are within one tenant, so with tenant_id this problem mostly disappears.
2. Let a database that supports it do the work. Distributed SQL databases, which look like one SQL database but shard and replicate underneath (CockroachDB, Spanner, YugabyteDB) and Citus run cross-shard transactions for you, using two-phase commitTwo-phase commit (2PC)A protocol for committing one transaction across several databases: a coordinator first asks everyone to “prepare” (promise they can commit), then tells all of them to commit. Correct but slow, and stuck if the coordinator dies mid-way. under the hood. They're slower than single-shard ones, but correct.
3. Use a SagaSagaA long business process split into steps, each with its own transaction, plus an “undo” step (a compensation) for each one. If step 3 fails, the compensations for steps 2 and 1 run instead of a rollback.. Split the operation into steps, each a local transaction, with an "undo" step for each (refund Alice if crediting Bob fails). The data is briefly inconsistent, but it heals. See the saga topic.

Resharding: adding machines without reshuffling everything

Why hash % N is a disaster

Load Balancing §05 worked this out for caches: when N goes from 4 to 5, a key keeps its server only if h % 4 happens to equal h % 5, which is true for about 1 key in 5. So about 80% of keys change server.

For a cache, that's a burst of misses. For a database it's far worse: 80% of your rows have to be copied over the network to a new machine, while the app keeps reading and writing them, and a router has to know for every single key whether it has moved yet. That's Resharding / rebalancingResharding / rebalancingMoving data between shards, usually because you added machines or one shard got too big or too busy, ideally while the app keeps running. at its most painful.

Fix 1: many fixed partitions

Split the formula into two steps, as our router in §05 did:

1. partition = hash(key) % 1024, where 1,024 never changes.
2. A table says which node owns each partition.

With 4 nodes, each owns 256 partitions. Add a fifth and it takes about 205 partitions from the others (about 51 from each). Only those partitions move, each as one whole unit, and the formula itself is untouched. Try it:

This is the most common design in production. Redis Cluster has 16,384 hash slots; Couchbase has 1,024 vBuckets; Elasticsearch fixes the number of primary shards when you create an index (Elasticsearch's word for a table, not the lookup structure from §07). The one decision you must get right up front is how many partitions: it's the most nodes you can ever spread across. A common rule of thumb is 10–100× the number of nodes you expect, since each partition has a small fixed overhead.

Fix 2: consistent hashing

Consistent hashingConsistent hashingA way to map keys (user IDs, cache keys) to servers so that adding or removing one server moves only about 1/N of the keys, instead of reshuffling almost all of them as hash(key) % N does. puts nodes and keys on a ring, and each key belongs to the next node clockwise. Adding a node takes over only the keys just before it, about 1/N of the total. With many virtual nodesVirtual node (vnode)One of many points a single server gets on a consistent-hash ring. With ~100 points per server, each server's share of keys evens out, and a removed server's keys spread over all the others. per server, the load stays even. Cassandra and DynamoDB are built this way.

You already know the details: Load Balancing §05 has the interactive ring and the code. Notice that "many virtual nodes per server" and "many fixed partitions per server" are the same idea in two outfits: cut the key space into many more pieces than you have machines, and move pieces.

Fix 3: split ranges as they grow

Range-partitioned systems don't pick a number up front. They start with one range covering everything and split any range that gets too big (or too busy), then move halves to emptier nodes:

Bigtable and HBase call these pieces regions or tablets; CockroachDB calls them ranges and splits them at about 512 MiB by default in recent versions; MongoDB calls them chunks. The upside: the system adapts to your data on its own, including hot ranges. The downside: a brand-new empty table starts as one range on one node, so the first burst of writes isn't spread until splits happen (many systems let you pre-split to avoid that).

Which one when

Moving data while the app keeps running

Why not just copy it?

The naive plan: stop writes, copy partition 7 from db-a to db-d, update the routing table, start writes again. For a few megabytes that's fine. For 200 GB the copy takes an hour, and "nobody can write for an hour" is an outage. The trick is to copy while writes continue, without losing any of them.

Step through it

1// writes for partition 7
2switch phase {
3case "normal":
4    oldDB.Write(row)
5case "dual": // backfill, verify
6    oldDB.Write(row) // the truth
7    newDB.Write(row) // keep up
8case "reads-new":
9    oldDB.Write(row)
10    newDB.Write(row)
11case "done":
12    newDB.Write(row)
13}

The pieces, named

• Dual writesDual writesDuring a migration, writing every change to both the old and the new location so the new one stays up to date. keep the new location current from the moment the move starts. The old location stays the source of truth until the cutover.
• The backfillBackfillA background job that copies existing rows from the old location to the new one in small batches, while new writes are handled separately. copies everything that existed before dual writes began. It must be IdempotentIdempotentDoing it twice has the same effect as doing it once. “Set balance to 50” is idempotent; “add 50 to balance” is not — so only the first is safe to retry blindly. (safe to re-run after a crash) and must never overwrite a newer version with an older one: insert-if-missing, or compare a version number or updated_at.
• Verify compares row counts and checksumsChecksumA small number computed from a block of data and sent along with it. The receiver recomputes it; if the two don't match, the data was damaged in transit. Also used to check that two copies of data are identical. per batch of rows, and re-copies any batch that differs.
• The cutoverCutoverThe moment during a migration when traffic switches from the old location to the new one. Good migrations make it a quick, reversible flip of a setting. is a change to the routing table, reads first, then writes. Each flip is reversible until the old copy is deleted.

The cutover moment

The stepper above flips writes in one clean step. Real life is messier, because every app server keeps its own copy of the routing table (§05), and the copies don't all update at the same instant. For a moment after the flip, some servers still send writes for p7 to db-a only, after db-d has become the truth. With change data capture the copy stream has already been stopped by then, so those writes are simply lost. Step through both versions:

The fix is a very short write pause, enforced at the old database: hold writes for that one partition, wait until the change stream has caught up completely, flip the routing table, then make the old database refuse writes for p7 so stale routers refresh and retry. Done well, it lasts milliseconds to a couple of seconds and only affects keys in the moving partition.

This same recipe (dual write or change stream → backfill → verify → cutover) is how teams move between databases entirely, change a shard key, or split a MicroserviceMicroserviceA small, separately deployed program that owns one part of a bigger system and talks to the others over the network (HTTP, gRPC, a message queue).'s data out of a shared database. Learn it once.

Postgres: partitions on one server, shards with Citus

Why partition on one machine at all?

• Dropping old data. Deleting 40 million old rows with DELETE is slow and leaves the table bloated (the disk space isn't given back right away). Dropping last year's partition is a quick file deletion.
• Skipping work. A query for October only reads the October partition (Partition pruningPartition pruningThe database skipping partitions that can't contain matching rows, by comparing the query's WHERE clause with each partition's range. A query for October reads only the October partition.).
• Smaller pieces. Each partition has its own, smaller indexes, and Postgres's background clean-up (VACUUM) runs per partition.

It's also the step many teams take before sharding (§11), and the same table design carries over when they do shard.

Range partitioning by month

This uses Postgres's declarative partitioningDeclarative partitioning (Postgres)Postgres's built-in way to split one table into partitions on the same server: you declare PARTITION BY RANGE / LIST / HASH, and Postgres routes rows and prunes partitions for you.. Everything below ran on PostgreSQL 16; the output is pasted unchanged.

CREATE TABLE orders (
  id         bigint      NOT NULL,
  created_at timestamptz NOT NULL,
  total      numeric     NOT NULL
) PARTITION BY RANGE (created_at);           -- ← the partition key

CREATE TABLE orders_2026_08 PARTITION OF orders
  FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');
CREATE TABLE orders_2026_09 PARTITION OF orders
  FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');
CREATE TABLE orders_2026_10 PARTITION OF orders
  FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');

We inserted 130,000 orders (one per minute from 1 August) into the parent table. Postgres routed each row to the right partition by itself. tableoid::regclass shows which partition a row is physically in (x::type converts x to another type; here an internal table number becomes a readable table name, and GROUP BY 1 ORDER BY 1 means "group and sort by the first selected column"):

SELECT tableoid::regclass AS partition, count(*) FROM orders GROUP BY 1 ORDER BY 1;

   partition    | count
----------------+-------
 orders_2026_08 | 44639
 orders_2026_09 | 43200
 orders_2026_10 | 42161

Partition pruning in EXPLAIN

EXPLAIN shows the plan Postgres would use for a query, without running it (the indexes topic teaches how to read plans; COSTS OFF just hides the cost estimates). Ask for one week of October:

EXPLAIN (COSTS OFF)
SELECT sum(total) FROM orders WHERE created_at >= '2026-10-01' AND created_at < '2026-10-08';

 Aggregate
   ->  Seq Scan on orders_2026_10 orders
         Filter: ((created_at >= '2026-10-01 00:00:00+00'::timestamp with time zone) AND (created_at < '2026-10-08 00:00:00+00'::timestamp with time zone))

Read plans from the inside out: Seq Scan (a Seq ScanSeq Scan (sequential scan)A query-plan step that reads every row of a table or partition in order, without using an index.) reads one table or partition row by row, Filter keeps the matching rows, and Aggregate computes the sum. Only one partition appears. August and September were pruned because their ranges can't contain October rows. Now filter by something that isn't the partition key:

EXPLAIN (COSTS OFF) SELECT sum(total) FROM orders WHERE id = 42;

 Aggregate
   ->  Append
         ->  Seq Scan on orders_2026_08 orders_1
               Filter: (id = 42)
         ->  Seq Scan on orders_2026_09 orders_2
               Filter: (id = 42)
         ->  Seq Scan on orders_2026_10 orders_3
               Filter: (id = 42)

Every partition is scanned and the results are glued together (Append = combine the output of several scans). That's scatter-gather (§07) in miniature, on one machine: queries that don't filter by the partition key visit every partition.

Pruning also works with query parameters ($1), the way Go and Node send them. Postgres then prunes when the query runs; with EXPLAIN ANALYZE you see Subplans Removed: 1 instead of the skipped partitions.

Two errors you will meet

Insert a row that no partition accepts:

ERROR:  no partition of relation "orders" found for row
DETAIL:  Partition key of the failing row contains (created_at) = (2026-12-25 00:00:00+00).

Create next month's partition ahead of time (a scheduled job, or the pg_partman extension), or add a catch-all with CREATE TABLE orders_default PARTITION OF orders DEFAULT;.

Now add a Primary keyPrimary keyThe column (or columns) that uniquely identifies each row, like users.id. The database refuses two rows with the same primary key and indexes it automatically. on id:

ERROR:  unique constraint on partitioned table must include all partitioning columns
DETAIL:  PRIMARY KEY constraint on table "orders" lacks column "created_at" which is part of the partition key.

Postgres only checks uniqueness inside each partition, so it can only guarantee a unique key that includes the partition column: PRIMARY KEY (id, created_at) works. Sharded databases have the same limit at a bigger scale: a unique constraint is only enforced per shard. That's one reason globally unique IDs must come from the ID generator itself (§11).

Dropping a month

ALTER TABLE orders DETACH PARTITION orders_2026_08;   -- no longer part of orders
DROP TABLE orders_2026_08;                            -- 44,639 rows gone, instantly

Hash partitioning

CREATE TABLE events (tenant_id bigint NOT NULL, id bigint NOT NULL, body text)
  PARTITION BY HASH (tenant_id);
CREATE TABLE events_p0 PARTITION OF events FOR VALUES WITH (MODULUS 4, REMAINDER 0);
-- … and the same for REMAINDER 1, 2, 3

MODULUS 4, REMAINDER 0 means "rows where Postgres's hash of tenant_id, divided by 4, leaves remainder 0". With 100,000 rows for 1,000 tenants we got 25,900 / 23,400 / 27,600 / 23,100: roughly even, as hashing should be. A query for tenant_id = 7 was pruned to one partition (events_p3); tenant_id BETWEEN 7 AND 9 scanned all four, because hashing destroys order (§03).

Across servers: Citus

CitusCitusA Postgres extension that spreads tables across many Postgres servers. A coordinator node routes queries and merges results. is a Postgres extension that turns a group of Postgres servers into one sharded database. You create a normal table, then tell Citus which column to distribute by:

SELECT create_distributed_table('events', 'tenant_id');
SELECT create_distributed_table('projects', 'tenant_id', colocate_with => 'events');

Citus hash-partitions the table into many shards (32 by default in recent versions) spread over the worker servers, and a coordinator routes each query. colocate_with puts the same tenants' rows of both tables on the same workers, so JOINs on tenant_id stay local (§04's co-location). Cross-shard transactions use two-phase commit (§07). We didn't run this one (Citus isn't installed in our sandbox), so treat the exact defaults as "check your version's docs".

Staff-level: IDs, load, cost, and when not to shard

1. IDs: auto-increment breaks

On one database, id bigserial (Postgres's auto-increment 64-bit column) hands out 1, 2, 3… On four shards, each shard counts on its own, so there are four order #1s. And as §10 showed, a unique constraint is only checked within a shard. You need IDs that machines can create independently without colliding. Two good options:

UUIDv7UUIDv7A UUID whose first 48 bits are the current time in milliseconds, followed by random bits. New IDs sort roughly by creation time, which keeps B-tree indexes compact. is a 128-bit UUIDUUID (v4 / v7)A 128-bit ID like 3f2b…-…. Version 4 is fully random; version 7 starts with a timestamp, so new IDs are roughly in increasing order. that starts with the current time in milliseconds, followed by random bits. Because new IDs are roughly in time order, they're inserted at the "end" of a B-treeB-treeA wide, shallow, always-sorted tree that most databases use for tables and indexes. Each node fills one disk page and holds hundreds of keys, so finding one row among millions takes only 3–4 page reads. index, like auto-increment IDs, instead of at random places (which is what makes the older, fully random UUIDv4 slow for big indexes). PostgreSQL 18 adds a built-in uuidv7() function; in Go, github.com/google/uuid has uuid.NewV7(); in Node, the uuid package exports v7(). (Node's built-in crypto.randomUUID() makes v4.)

A Snowflake IDSnowflake IDA 64-bit ID made of a timestamp, a machine number and a per-millisecond counter, invented at Twitter. Unique without coordination, sortable by time, and half the size of a UUID. packs a timestamp, a machine number and a counter into 64 bits, half the size of a UUID:

const epoch = 1767225600000 // 2026-01-01 00:00 UTC, in milliseconds

func NewIDGen(node int64) (*IDGen, error) {
	if node < 0 || node > 1023 { // more than 10 bits would spill into the time bits
		return nil, fmt.Errorf("node %d out of range 0…1023", node)
	}
	return &IDGen{node: node}, nil
}

type IDGen struct {
	mu     sync.Mutex
	node   int64 // 0…1023, unique per running process
	lastMs int64
	seq    int64 // 0…4095 within one millisecond
}

func (g *IDGen) Next() int64 {
	g.mu.Lock()
	defer g.mu.Unlock()
	now := max(time.Now().UnixMilli(), g.lastMs) // never go back in time
	if now == g.lastMs {
		g.seq = (g.seq + 1) & 4095 // keep the low 12 bits: 4095 + 1 wraps to 0
		if g.seq == 0 {            // 4,096 IDs this millisecond already: wait for the next one
			for now <= g.lastMs {
				now = time.Now().UnixMilli()
			}
		}
	} else {
		g.seq = 0
	}
	g.lastMs = now
	return (now-epoch)<<22 | g.node<<12 | g.seq // time | node | sequence
}

We ran both: consecutive IDs come out increasing (109617025256550400, …401), and decoding the node bits gives back 7. Our JS version keeps its state in module variables, so run one generator per process, each with a different node number.

Two honest limits of this tiny version. Both check that node fits in 10 bits, because a bigger number would silently overwrite the time bits. And if the server's clock jumps backwards by several seconds (it happens when clocks get corrected), max(…, lastMs) keeps IDs unique, but the "wait for the next millisecond" loop can spin until the clock catches up, while holding the mutex. Production generators log loudly or refuse to issue IDs when they see the clock go back.

2. Balance load, not just bytes

Partitions are rarely equal. One tenant runs a nightly import; one region wakes up at 9 a.m. So:

• Measure per partition: requests per second, CPU, p99, size. "Shard 7 is at 92% CPU while the average is 40%" is the alert that matters.
• Move hot partitions to quiet nodes, or split them (§08). Moving the biggest partition isn't always right; move the busiest per byte moved.
• Leave headroom. Run each node at well below its limit (often around 50–60%) so a FailoverFailoverSwitching to a standby (a replica) when the primary server fails, so the system keeps working — automatically or by an operator. or a viral tenant has room to land (HeadroomHeadroomSpare capacity you keep on purpose — e.g. workers that are only 80% busy on average — so random bursts can be cleared quickly instead of piling up.).

3. The N× operational bill

4. When not to shard

Run through this list first. Each item is much cheaper than sharding, and together they carry most companies a very long way:

1. Fix the queries. Missing indexes and N+1 query loops (indexes, N+1) often hide a 10× win.
2. Scale up. A single modern server can have hundreds of CPU cores, terabytes of RAM and tens of terabytes of fast SSD. It's boring and it works.
3. Read replicas and caching for read-heavy load.
4. Partition on one server (§10) for retention and pruning.
5. Split by feature. Put billing, search and analytics in their own databases. Each is smaller, and nobody needs cross-database JOINs between them.
6. Archive cold data to cheaper storage.

Shard when writes or data size for one logical dataset really exceed what one machine (plus all of the above) can handle, or when you can see that point coming within a year or so. And if you can, consider a database that shards for you (CockroachDB, Spanner, YugabyteDB, Vitess, Citus, DynamoDB, Cassandra) before hand-rolling the router, the rebalancer and the migration tooling yourself.

Pitfalls gallery

Check yourself

Predict the answer before you click. Q1–Q2 are covered in §03, Q3–Q4 in §08, Q5 in §07, Q6 in §06, Q7 in §04, Q8 in §11 and Q9 in §10.

orders is range-partitioned by created_at, one month per shard:
  shard 0: Aug   shard 1: Sep   shard 2: Oct   shard 3: Nov
It's 28 November and traffic is 10× normal.

users is hash-partitioned by user_id over 8 shards.
SELECT * FROM users
WHERE signed_up_at BETWEEN '2026-01-01' AND '2026-03-31';

Each shard: 99% of answers in 10 ms, 1% take 200 ms (independently).
A dashboard query asks all 50 shards and waits for every answer.

Every request is for one customer company (tenant).
Most screens show one tenant's projects, tasks and users together.

// the API returns: {"id": 9007199254740993}
const order = await res.json();
console.log(order.id);

CREATE TABLE orders (id bigint, created_at timestamptz, ...)
  PARTITION BY RANGE (created_at);
ALTER TABLE orders ADD PRIMARY KEY (id);

Cheat sheet