# Partitioning & sharding > Split data across machines: choose a partition key, compare range and hash partitioning, rebalance without moving everything, and handle hot keys, secondary indexes and cross-shard queries. System Design roadmap · Data at scale · 20 min read · https://levelcode.org/roadmaps/system-design/partitioning Replication copies the same data onto several machines. It scales reads but not writes, and every replica still has to hold the whole dataset. When the write rate or the data size outgrows the largest machine you can reasonably buy, you **partition**: split the data so each machine holds and serves only a slice. (Databases call the slices partitions, shards, regions, tablets or vnodes; they all mean the same thing.) Partitioning is powerful and expensive. This lesson covers the decisions that determine whether it goes well: the key, the scheme, rebalancing, hot spots, and queries that span shards. ## The partition key decides everything Every record gets a **partition key**, and a function maps keys to partitions. Choose the key from your access patterns, because the goal is for the common queries to touch **one partition**: | Workload | Good key | Why | |---|---|---| | Multi-tenant SaaS | `tenant_id` | Almost every query is scoped to one tenant | | Chat | `conversation_id` | Messages are read and written per conversation | | Social graph | `user_id` | "My followers", "my timeline" are per user | | Time-series metrics | `(metric_id, time bucket)` | Spreads writes while keeping a series' recent data together | A key that queries don't use forces every query to ask every partition — the worst outcome of partitioning. And a key that sends most traffic to a few values creates **hot partitions** no matter how many machines you add. ## Range partitioning Assign each partition a contiguous range of keys: A–F on shard 1, G–M on shard 2, and so on. HBase, Bigtable, CockroachDB, TiDB and MongoDB's ranged sharding work this way. - **Range queries are efficient.** "All orders for customer 4,200–4,300" or "events from 10:00 to 10:05" touch one or two partitions, and keys within a partition are stored sorted. - **Boundaries adapt to data.** When a partition grows too large, split it in two; when two shrink, merge them. - **Sequential keys create a hot spot.** If the key is a timestamp or an auto-increment ID, every new write lands in the last range, so one partition takes all the writes while the rest sit idle. Prefix the key with something that spreads writes (`sensor_id` before the timestamp) or use hash partitioning. ## Hash partitioning Hash the key and assign partitions by hash value. A good hash scatters similar keys uniformly, so load spreads evenly even when the keys themselves are sequential. Cassandra, DynamoDB and Redis Cluster all hash. The cost: **range queries are gone.** Adjacent keys land on different partitions, so "orders between two dates" must ask all of them. A common compromise is a **compound key**: hash the first part to choose the partition, and sort by the second part within it. In Cassandra, `PRIMARY KEY ((user_id), created_at)` puts all of a user's rows on one partition, sorted by time, so "this user's latest 20 posts" is a single sequential read. See [range vs hash partitioning](https://levelcode.org/concepts/sharding). ## Rebalancing without moving everything Partitions must move when you add or remove machines. The naive assignment, `partition = hash(key) mod N`, is a disaster here: change N from 4 to 5 and about 80% of keys map to a different node, so nearly the whole dataset moves at once. Two designs fix it. **Consistent hashing** puts nodes and keys on a ring; each key belongs to the next node clockwise. Adding a node takes keys only from its neighbour, so about 1/N of the data moves. Virtual nodes (each machine placed on the ring many times) even out the shares. The DevSecOps load-balancing lesson works through the ring step by step; try it here. See [consistent hashing](https://levelcode.org/concepts/consistent-hashing). > **Interactive: Modulo versus consistent hashing.** Replace 'backend' with 'shard'. Under modulo hashing, removing one node remaps most keys; on the ring only the removed node's keys move. Try it: https://levelcode.org/labs/load-balancer **A fixed number of partitions** is simpler and more common in practice. Create many more partitions than nodes up front — Redis Cluster has 16,384 hash slots, Elasticsearch indexes have a fixed shard count, Kafka topics a fixed partition count — and assign several partitions to each node. Adding a node means moving *whole partitions* to it; the key-to-partition mapping never changes, only the partition-to-node table does. ```text slot = CRC16(key) mod 16384 # Redis Cluster node = slot_table[slot] # e.g. slots 0–5460 → A, 5461–10922 → B, 10923–16383 → C ``` Choose the partition count for your largest future size: too few and each partition becomes huge and slow to move; too many and per-partition overhead (files, metadata, open connections) adds up. > [!TIP] > Rebalance gradually and on purpose. Moving partitions copies a lot of data and competes with user traffic. Most databases let you trigger rebalancing manually or throttle it, which is safer than fully automatic moves set off by a node that's merely slow. ## Routing requests to the right partition Something has to know which node holds which partition. Three common designs: - **Any node routes.** The client talks to any node, which forwards to the owner (Cassandra coordinators). - **A routing tier.** A proxy layer knows the map and forwards (MongoDB's `mongos`, Vitess's VTGate). - **Smart clients.** The client library caches the map and talks to the owner directly; a node that receives a misdirected request replies with a redirect (Redis Cluster's `MOVED`). The authoritative map usually lives in a small consensus-backed store — ZooKeeper, etcd, or the database's own metadata service — so every router agrees on it. See [consensus and Raft](https://levelcode.org/concepts/consensus-raft). ## Hot partitions and hot keys Even with perfect hashing, one *key* can be hot: a celebrity's profile, a viral post, a single big customer in a multi-tenant system. All requests for that key go to one partition, and adding machines does nothing. See [hot partitions](https://levelcode.org/concepts/hot-partitions). Defences, from lightest to heaviest: 1. **Cache it.** Hot reads are usually the easiest case; a cache (even an in-process one) absorbs them. 2. **Salt the key for writes.** Write to `post:42#0` … `post:42#9`, chosen at random, so ten partitions share the load. Reads must fetch all ten and combine them — fine for counters (sum them), awkward for anything else. 3. **Give big tenants their own partition** or even their own cluster. Many SaaS systems move their largest customers to dedicated shards. 4. **Split by a second key** for the hot entity (for a huge conversation, partition by `(conversation_id, month)`). Detect hot keys before users do: track per-partition request rates and alert when one is far above the median. ## Secondary indexes across partitions Partitioning by `user_id` makes "this user's orders" fast. But how do you find "all orders with status `failed`"? Status isn't the partition key. Two options, each expensive in a different place. See [secondary indexes on sharded data](https://levelcode.org/concepts/secondary-indexes-partitioned). **Local (document-partitioned) index.** Each partition indexes only its own rows. Writes are cheap — one partition — but a query by status must ask *every* partition and merge the results (**scatter-gather**). Latency is set by the slowest partition, and the query load multiplies by the partition count. **Global (term-partitioned) index.** The index itself is partitioned by the indexed value: all `failed` entries on one index partition. Reads hit one index partition, but each write must update the data partition *and* one or more index partitions, usually asynchronously — so the index lags the data. DynamoDB's global secondary indexes work this way and are eventually consistent. If a query isn't on the hot path, scatter-gather over a local index is often fine. For hot queries on another key, a global index — or a second copy of the data partitioned by that key — is the usual answer. ## Cross-partition queries and transactions Joins and transactions that span partitions are where partitioned systems get hard: - **Design so they're rare.** Choose the key so related data lives together (a tenant's users, projects and tasks all keyed by `tenant_id`). - **Denormalise** data that's read together, accepting that copies must be kept in sync ([normalization vs denormalization](https://levelcode.org/concepts/normalization-vs-denormalization)). - **Use a database that does distributed transactions** (Spanner, CockroachDB, YugabyteDB) when you genuinely need them, and accept the extra latency of a commit protocol across nodes. - **Use sagas** for multi-step business operations across services, with compensating actions instead of a single atomic commit ([sagas](https://levelcode.org/concepts/saga-pattern)). Before any of this, check whether you need partitioning at all. A single PostgreSQL primary on a large machine handles terabytes and thousands of writes per second; partitioning a database that would have fit on one machine adds every problem in this lesson and solves none. ## Practice You're partitioning the database of a multi-tenant invoicing product: 40,000 tenants, but the largest 20 account for 35% of all rows and traffic. Queries: list a tenant's invoices by date; fetch one invoice by ID; a nightly job finds every overdue invoice across all tenants; and support staff search invoices by customer email across tenants. Choose a partition key and scheme, then explain how you'd handle the big tenants, the overdue job, and the support search. **Solution** **Key and scheme.** Partition by `tenant_id`, because nearly every query is scoped to one tenant and keeping a tenant's data together keeps those queries — and transactions — single-partition. Use hash partitioning over a fixed number of partitions (for example 1,024 logical shards mapped to a handful of physical nodes) so tenants spread evenly and nodes can be added by moving whole shards. Within a shard, index `(tenant_id, issued_at)` for "invoices by date", and make invoice IDs carry or map to the tenant (for example a `(tenant_id, invoice_id)` primary key, with the tenant in the URL) so fetching by ID is a single-shard lookup. **Big tenants.** The top 20 would make their shards hot. Give each of them a dedicated shard (or a dedicated node) by recording an override in the shard map, and move them there gradually. If a single tenant ever outgrows a node, sub-partition it by `(tenant_id, year)`. **Overdue job.** It's a cross-tenant batch, not a user request, so scatter-gather is acceptable: run it per shard in parallel, each using a local partial index such as `ON invoices (due_date) WHERE status = 'unpaid'`, and throttle it so it doesn't compete with daytime traffic. **Support search by email.** Scatter-gathering every shard on each search is slow and loads every node. Maintain a global lookup instead: a separate table or search index partitioned by email (`email → [(tenant_id, invoice_id)]`), updated asynchronously from invoice changes (for example through an outbox and a consumer). It lags by seconds, which support can live with. ## Key takeaways - Partition when writes or data size outgrow one machine; replicas scale only reads. - Pick the partition key from access patterns so common queries touch one partition. - Range partitioning keeps range scans fast but makes sequential keys hot; hash partitioning spreads load but scatters ranges. Compound keys combine them. - Never use `hash mod N` for placement: use consistent hashing or many fixed partitions moved whole. - Hot keys need caching, salting or dedicated partitions; more machines don't help. - Secondary indexes are either local (cheap writes, scatter-gather reads) or global (cheap reads, slower and lagging writes). ## Quiz 1. An events table is range-partitioned by event timestamp. What goes wrong under heavy ingestion? A. Range queries become impossible B. Every new write falls into the newest range, so one partition takes all the write load C. Rows are spread too evenly to find D. The database must rehash every key on each insert **Answer: B.** Sequential keys all land at the end of the key space. Prefixing with a well-distributed value (a source ID) or hashing the key spreads the writes. 2. Why do Redis Cluster and Kafka use a fixed number of partitions (slots) much larger than the node count? A. Adding a node only moves whole partitions; the key-to-partition mapping never changes B. More partitions always make queries faster C. It allows range queries across nodes D. It removes the need for replication **Answer: A.** Keys map to partitions with a fixed function. Rebalancing then just reassigns some partitions to the new node, moving roughly 1/N of the data instead of rehashing everything. 3. Orders are partitioned by user_id with a local index on status. What does a query for all orders with status = 'failed' cost? A. One partition lookup B. Nothing: local indexes are global C. A scatter-gather to every partition, with latency set by the slowest D. It's impossible without a full table rewrite **Answer: C.** Each partition indexes only its own rows, so a query not on the partition key must ask all of them and merge. A global index would make the read single-partition at the cost of more complex, lagging writes. 4. One celebrity's like counter gets 200,000 increments per second, overwhelming its partition. Which fix addresses it? A. Add more nodes to the cluster B. Switch from hash to range partitioning C. Split the counter across several salted keys and sum them on read D. Increase the replication factor **Answer: C.** All writes for one key go to one partition, so more nodes or replicas don't help. Salting spreads the key over several partitions; for a counter, reads simply add the parts.