# Partitioning and Sharding Plan: [Dataset or service]

> Building-block decision record for splitting a dataset across machines.
> It covers the partition key, the partitioning scheme (hash, range,
> consistent hashing, directory), partition count and placement,
> rebalancing, hot keys, cross-partition queries and the migration plan.
> Concepts draw on consistent hashing (Karger et al., STOC 1997), on the
> Dynamo paper (DeCandia et al., SOSP 2007) for virtual nodes and
> preference lists, on rendezvous hashing (Thaler and Ravishankar, 1998) and
> on jump consistent hashing (Lamping and Veach, 2014). Replace every
> `[bracketed]` value and every row marked *Example*. Worked examples use
> Quillmere, an invented book-review and reading-list platform shared by
> the other building-block templates in this category.

<!--
Guidance: How to use this template
- Choose the partition key from the access patterns, not from the schema.
  List the top queries first (section 2); the key is the field that most
  of them can route on.
- Decide the number of logical partitions once and make it much larger
  than the number of machines. Moving whole partitions between machines
  is easy; changing how keys map to partitions is a migration.
- Estimate skew before launch: a perfect hash spreads keys evenly, but not
  load. One popular key can still overload one partition (section 6).
- For the full design of a partitioned store (replication, anti-entropy,
  storage engine) use the Distributed Key-Value Store System Design
  template (`system-design/design-distributed-kv-store`).
-->

## Document control

<!-- Guidance: Partitioning is hard to change later; get sign-off from the teams that own the main queries, not only from the storage team. -->

| Field | Value |
|---|---|
| Document ID | [SHARD-REC-001] |
| Dataset | [*Example:* Reading-list entries (user, book, shelf, progress, timestamps)] |
| Store | [*Example:* Self-managed relational cluster, one database per shard] |
| Current size and growth | [*Example:* 12 TB, growing 25% per year] |
| Version | [0.1] |
| Status | [Draft / In review / Approved / Implemented / Superseded] |
| Owner | [@data-platform-lead] |
| Reviewers | [@reading-list-lead], [@sre], [@dba] |
| Last updated | YYYY-MM-DD |
| Classification | [Internal] |
| Related | Distributed Key-Value Store System Design (`system-design/design-distributed-kv-store`), Replication and Consistency Models (`system-design/replication-consistency-models`), Database Design (`architecture/database-design`) |

### Approvals

| Role | Name | Decision | Date |
|---|---|---|---|
| Data platform | [@name] | [Approved / Changes requested] | YYYY-MM-DD |
| Main consumer team | [@name] | [Decision] | YYYY-MM-DD |
| SRE | [@name] | [Decision] | YYYY-MM-DD |

### Revision history

| Version | Date | Author | Change summary |
|---|---|---|---|
| 0.1 | YYYY-MM-DD | [@name] | First draft |
| 0.2 | YYYY-MM-DD | [@name] | *Example:* Added salted keys for the "trending shelf" hot key |

## 1. Why partition

<!-- Guidance: Name the limit you are hitting: storage per node, write throughput, memory for indexes, or blast radius. If reads are the only problem, replicas or a cache may be cheaper than sharding. -->

[Reason. *Example:* The single primary holds 12 TB and accepts about
18,000 writes per second at peak; index maintenance and backups no longer
fit the maintenance window, and a primary failure affects every user.
Read replicas do not help with writes.]

| Limit | Current | Ceiling of one node | Headroom |
|---|---|---|---|
| *Example:* Storage | 12 TB | 16 TB usable | About 1 year at 25% growth |
| *Example:* Writes | 18,000 per second | 20,000 per second at p99 target | Under 6 months at 25% growth (about 5.7) |
| [Limit] | [Value] | [Value] | [Time] |

## 2. Access patterns and partition key

<!-- Guidance: List every important query with its rate and whether it can name the key. A query that cannot name the key becomes a scatter-gather across all partitions. -->

| ID | Query | Share of traffic | Includes `user_id`? | Includes `book_id`? |
|---|---|---|---|---|
| Q-01 | *Example:* Get a user's shelves and entries | 61% | Yes | No |
| Q-02 | *Example:* Update reading progress | 27% | Yes | Yes |
| Q-03 | *Example:* Count readers of a book | 9% | No | Yes |
| Q-04 | *Example:* Admin export for one user | 3% | Yes | No |
| [Q-NN] | [Query] | [Share] | [Yes/No] | [Yes/No] |

Partition key: [*Example:* `user_id`. It serves Q-01, Q-02 and Q-04 (91%
of traffic) on one partition. Q-03 moves to a separate counter maintained
from a change stream, so it never scatters.]

Checklist for the key:

- [ ] High cardinality (many more distinct values than partitions).
- [ ] Stable: the value never changes for a record, or a change is handled
      as a delete and insert across partitions.
- [ ] Spreads load, not only data: no single value carries a large share
      of requests.
- [ ] Transactions that must be atomic stay within one key value.

## 3. Partitioning scheme

<!-- Guidance: Compare the schemes for this dataset and record why one was chosen. Hash spreads load but loses range scans; range keeps order but invites hot spots on sequential keys. -->

| Scheme | How keys map | Strengths | Weaknesses |
|---|---|---|---|
| Hash modulo N | `hash(key) mod N` nodes | Simple, even spread | Changing N remaps almost every key (see below) |
| Fixed logical partitions | `hash(key) mod P`, P fixed and large, partitions assigned to nodes by a map | Moves whole partitions; N can change freely | P must be chosen up front; very large P adds metadata |
| Consistent hashing ring | Keys and nodes hashed onto a ring, key goes to next node clockwise (Karger et al.) | Adding a node moves about 1/(N+1) of keys | Uneven with few points per node; needs virtual nodes |
| Rendezvous (highest random weight) | Each key picks the node with the highest `hash(key, node)` | No ring state, minimal movement | Lookup cost grows with node count |
| Jump consistent hash | Arithmetic function of key and bucket count | No memory, even spread | Buckets can only be added or removed at the end |
| Range | Contiguous key ranges per partition, split when large | Range scans, ordered reads | Sequential keys (timestamps, increasing IDs) hit one partition |
| Directory | Lookup table from key or tenant to partition | Full control, can move single tenants | Directory is a critical dependency and must be cached |

**Why modulo N fails on growth, *Example*:** in a simulation of 200,000
keys hashed with MD5, going from 8 to 9 nodes with `hash mod N` remapped
88.8% of keys, matching the expected 1 - 1/9 = 88.9%. A consistent hashing
ring with 128 virtual nodes per node remapped 11.6%, close to the ideal
1/9 = 11.1%.

**Virtual nodes and balance, *Example*** (same simulation, 8 nodes):

| Points per node on the ring | Busiest node vs mean | Lightest node vs mean | Coefficient of variation |
|---|---|---|---|
| 1 | 2.68 | 0.12 | 0.85 |
| 16 | 1.59 | 0.72 | 0.32 |
| 128 | 1.08 | 0.93 | 0.05 |

Decision: [*Example:* Fixed logical partitions. P = 256, assigned to
nodes by a versioned partition map held in the coordination service.
`partition = hash(user_id) mod 256`.]

```mermaid
flowchart LR
  R["Request with user_id"] --> H["hash(user_id) mod 256"]
  H --> M["Partition map v42"]
  M --> N1["Node A: partitions 0 to 31"]
  M --> N2["Node B: partitions 32 to 63"]
  M --> N8["Node H: partitions 224 to 255"]
```

## 4. Partition count and placement

<!-- Guidance: Size partitions so that one can be copied in minutes, not hours, and so that the count divides evenly across likely node counts. Record the placement rules (zones, racks) for replicas. -->

| Input | *Example* value |
|---|---|
| Data today | 12 TB |
| Target partition size | 50 GB, so at least 12,000 / 50 = 240 partitions today |
| Logical partitions (P) | 256 (next power of two above 240, leaves room for growth) |
| Nodes at launch | 8, so 256 / 8 = 32 partitions per node |
| Replicas per partition | 3, one per availability zone |
| Placement rule | No two replicas of a partition in the same zone; no node holds more than 34 primaries |

As data grows past about 50 GB per partition (256 x 50 GB = 12.8 TB),
plan either a larger P through a split (each partition into two, by one
more hash bit) or a larger target size, decided before the threshold.

## 5. Rebalancing

<!-- Guidance: Describe how a partition moves while it keeps serving traffic. The usual sequence is copy a snapshot, stream changes, switch ownership in the map, then drain. Rate-limit copies so rebalancing does not cause the outage it is meant to prevent. -->

Adding a ninth node, *Example:* with fixed partitions the new node takes
about 256 / 9 = 28.4 partitions (28 or 29), the share 1/9 = 11.1% of the
data, and no key changes partition.

```mermaid
sequenceDiagram
  participant Ctl as Rebalancer
  participant Src as Source node
  participant Dst as New node
  participant Map as Partition map
  Ctl->>Dst: Copy snapshot of partition 17
  Src-->>Dst: Stream changes since snapshot
  Ctl->>Map: Set owner of 17 to new node (v43)
  Note over Src: Rejects writes for 17 with a moved reply and map v43
  Ctl->>Src: Delete partition 17 after grace period
```

| Rule | *Example* decision |
|---|---|
| Trigger | Manual approval, or automatic when a node exceeds 80% storage |
| Copy rate limit | 100 MB per second per node, at most 2 partitions moving per node |
| Ownership switch | Atomic map version bump; clients that use an old map get a "moved" reply and refresh |
| Rollback | Map version can be reverted while the source copy still exists |
| Verification | Row counts and checksums compared before the source is deleted |

## 6. Hot keys and skew

<!-- Guidance: Find the keys that carry the most load, not the most data. Options are splitting the key with a suffix, caching it, or giving it a dedicated partition. Each option moves complexity to readers. -->

*Example:* the global "trending shelf" counter receives 30,000 writes per
second, while one partition sustains about 5,000. It needs at least
30,000 / 5,000 = 6 sub-keys; the plan uses 8 (`trending:{0..7}`), so each
takes 3,750 writes per second (75% of capacity) and readers sum 8 values.

| Technique | When to use | Cost |
|---|---|---|
| Salted sub-keys | Hot write counters and append-only feeds | Reads fan out to all sub-keys |
| Cache in front | Hot reads (see `system-design/caching-strategy`) | Staleness |
| Dedicated partition or node | One very large tenant | Placement exceptions to manage |
| Time-bucketed keys | Sequential writes on a range scheme | Queries span buckets |
| Admission or rate limits | Abusive or accidental hot spots | Some requests rejected (see `system-design/design-rate-limiter`) |

Detection: [*Example:* per-partition request rate and top-20 keys sampled
every minute; alert when one partition exceeds 2 times the median.]

## 7. Cross-partition operations

<!-- Guidance: List every operation that spans partitions and how it is handled. Pitfall: secondary indexes. A local index per partition forces scatter-gather on reads; a global index needs its own partitioning and is updated asynchronously. -->

| Operation | Approach | Consistency | *Example* decision |
|---|---|---|---|
| Query by a non-key field | Global secondary index partitioned by that field, or scatter-gather | Index is eventually consistent | "Readers of a book" served from a counter table keyed by `book_id` |
| Multi-key transaction | Avoid by key design, or use a saga (`system-design/event-delivery-guarantees`) | Eventual, with compensation | Moving a book between two users' shelves is not supported |
| Unique constraint across partitions | Separate uniqueness table keyed by the unique value | Enforced at write time with a conditional insert | Username uniqueness in an `usernames` table |
| Global aggregates | Pre-computed from a change stream | Lagging by seconds | Daily active readers computed hourly |

## 8. Migration plan

<!-- Guidance: Moving from one node to many is a data migration. Use dual writes or a change stream with verification, and keep a rollback path until the old store is retired. Link to a data migration strategy where one exists. -->

| Step | Action | Exit criterion | Rollback |
|---|---|---|---|
| 1 | *Example:* Add routing layer that reads the partition map, all partitions on the old primary | Routing in production for 1 week, no errors | Remove routing layer |
| 2 | *Example:* Backfill 256 partitions to the new nodes from a snapshot and stream changes | Lag under 1 s for 24 hours | Stop stream |
| 3 | *Example:* Shadow reads compare old and new for 1% of traffic | Mismatch rate below 0.01% | Keep reading old |
| 4 | *Example:* Switch ownership in batches of 32 partitions | Error rate and latency unchanged per batch | Revert map version |
| 5 | *Example:* Retire old primary after 30 days | No reads for 30 days | Restore from final snapshot |

## 9. Decisions and open items

<!-- Guidance: Record decisions with reasons; link larger ones to an ADR (`architecture/adr-full`). -->

| ID | Decision or open item | Reason | Owner | Status |
|---|---|---|---|---|
| SD-01 | *Example:* Partition key `user_id` | Serves 91% of traffic on one partition | @data-platform-lead | Decided |
| SD-02 | *Example:* 256 fixed logical partitions | Moves without remapping; 32 per node at launch | @data-platform-lead | Decided |
| SD-03 | *Example:* 8 salted sub-keys for trending counter | 30,000 writes per second exceeds one partition | @reading-list-lead | Decided |
| SD-04 | *Example:* Plan for partition split at 12.8 TB | 50 GB target size | @dba | Open |
| [SD-NN] | [Item] | [Reason] | [@owner] | [Status] |

## Glossary and references

<!-- Guidance: Keep definitions short and consistent with the other building-block templates. -->

| Term | Meaning |
|---|---|
| Partition (shard) | Subset of the data stored and served as a unit |
| Partition key | Field whose value decides which partition holds a record |
| Virtual node | One of many points a physical node owns on a hash ring, used to even out load |
| Rebalancing | Moving partitions between nodes to spread data or load |
| Scatter-gather | Sending a query to every partition and merging results |
| Hot key | Key with a share of traffic large enough to overload its partition |

References:

- Karger D., Lehman E., Leighton T., Panigrahy R. et al., "Consistent
  hashing and random trees", Proceedings of the 29th Annual ACM Symposium
  on Theory of Computing (STOC), 1997, pages 654 to 663,
  https://doi.org/10.1145/258533.258660.
- DeCandia G. et al., "Dynamo: Amazon's highly available key-value store",
  Proceedings of the 21st ACM Symposium on Operating Systems Principles
  (SOSP), 2007, pages 205 to 220, https://doi.org/10.1145/1294261.1294281.
- Thaler D., Ravishankar C., "Using name-based mappings to increase hit
  rates", IEEE/ACM Transactions on Networking 6(1), 1998, pages 1 to 14,
  https://doi.org/10.1109/90.663936 (rendezvous hashing).
- Lamping J., Veach E., "A Fast, Minimal Memory, Consistent Hash
  Algorithm", arXiv:1406.2294, 2014, https://arxiv.org/abs/1406.2294.
- Related templates: Distributed Key-Value Store System Design
  (`system-design/design-distributed-kv-store`), Replication and
  Consistency Models (`system-design/replication-consistency-models`),
  Data Store Selection (`system-design/data-store-selection`), Caching
  Strategy (`system-design/caching-strategy`), Distributed Lock and
  Coordination System Design
  (`system-design/design-distributed-lock-coordination`).

## Standard and licence note

Based on the publicly available papers cited above; no standard governs
data partitioning. The simulation figures in section 3 come from
a simple model (MD5 hash of the strings `key-0` to `key-199999`, ring
points hashed from `node-<n>-vn-<v>`, one run) and illustrate the method
only; other key names give different single-run figures.
The text, tables and examples are original to this collection, and
Quillmere is an invented organisation. Check the documentation of your
database for its own partitioning and rebalancing features. Template
licence: MIT.