What I Learned About Partitioning: Data, Indexes, and Routing
Partitioning decides where data lives, which requests stay local, and how a cluster grows. My takeaway is to choose the layout around the workload, then account for indexing, movement, and routing.
Partitions, replicas, and nodes
Each record belongs to one logical partition, which can have replicas on several nodes. A node can lead some partitions and follow others. Replication provides copies; partitioning spreads storage and independent requests across machines. Both transactional and analytical systems use these techniques.
flowchart TB
subgraph N1[Node 1]
A[P1 leader]
B[P2 follower]
end
subgraph N2[Node 2]
C[P1 follower]
D[P2 leader]
end
A --> C
D --> B
💡 A shard is a partition; a node is a machine. Many partitions on one node give movement granularity, not extra hardware capacity. Tables can have separate maps or colocated keys. A network partition is different: a communication failure.
Key ranges preserve locality
A balanced layout spreads data and requests. Skew concentrates either; a hot spot receives disproportionate load. Random placement distributes records but makes lookups search everywhere unless a directory tracks them.
Range partitioning assigns contiguous sorted keys. Like encyclopedia volumes, ranges need not have equal widths: boundaries should follow the data, manually or automatically. The range map identifies a partition; its ownership map identifies a node.
flowchart TB
K[Sorted keys] --> A[P0: A to C]
K --> B[P1: D to M]
K --> C[P2: N to Z]
Q[Query D through F] --> B
Sorted keys support efficient range scans. Timestamp-first sensor keys, however, send every new reading to today's partition. Prefixing sensor ID spreads writers and preserves time scans per sensor; querying all sensors requires multiple scans.
Hashing trades order for distribution
A stable hash maps keys into numeric ranges owned by partitions. It must agree across processes; cryptographic strength is unnecessary, but process-dependent object hashes are unsafe for routing. Adjacent original keys become scattered, making original-key range scans expensive or unsupported.
flowchart LR
A[user41] --> B[Hash 82] --> C[P1: 50 to 99]
D[user42] --> E[Hash 17] --> F[P0: 0 to 49]
💡 Hashing distributes many keys statistically, not perfectly; sizes and popularity still matter. The consistent-hashing sidebar describes pseudorandom cache boundaries and limited remapping, unrelated to replica or ACID consistency. Remapped CDN entries can refill on a miss. Uneven ranges and durable-data movement make the original approach awkward for databases; later variants improve allocation or metadata costs.
Compound keys: in the Cassandra-style example, hash the user portion of (user_id, update_timestamp) and sort that user's records by timestamp. Fixing a user enables a local time-range scan; a global time scan or user-ID range loses that advantage.
A hot key needs special handling
Repeated writes to one celebrity or event hash to the same owner. Salt a known hot key into buckets such as event42:00 through event42:99. Writers choose a bucket; readers fetch and combine all buckets. These need not occupy 100 machines.
💡 Store which keys are split and their bucket counts so readers know what to fetch. Manual or automated configuration needs a migration rule when counts change. Salting adds fan-out and bookkeeping, so reserve it for hot keys. A size-triggered partition split cannot automatically spread one logical key.
Local secondary indexes
A secondary index finds records sharing an attribute rather than locating one primary key. In document-partitioned indexing, each partition maintains entries for its own documents. A write updates one document partition and its local indexes.
flowchart TB
Q[Search color:red] --> A[P0: red maps to 191]
Q --> B[P1: red maps to 768]
A --> R[Combine 191 and 768]
B --> R
Without a partition restriction, a coordinator scatters the search and gathers results. Each participant can use an index rather than scan documents, but the slowest response amplifies tail latency. Colocating common queries helps; multiple filters may need incompatible layouts.
💡 A one-partition query can use a local index when its partition key is known. This is not exclusively a NoSQL issue. A normal PostgreSQL connection does not search unrelated databases automatically; native table partitioning differs from distributed sharding. Declaring
CREATE INDEX users_city_idx ON users(city)alone does not specify distributed placement.
Global secondary indexes
A term-partitioned index covers documents across primary partitions and is itself partitioned. The owner of color:red can reference documents 191 and 768 wherever they live.
flowchart TB
A[Document 191 in P0] --> I[Index owner: color:red]
B[Document 768 in P1] --> I
Q[Search red] --> I
I --> R[IDs 191 and 768]
R --> F[Fetch from document owners]
Partition terms by value for range searches or hash for distribution. A single-term lookup can target one index partition; fetching documents or combining filters may still cross nodes. One write can update several index partitions. Synchronous consistency requires coordination; asynchronous indexing allows stale or missing search results.
💡 Application-built value-to-ID maps risk races and partial writes. Kafka or another queue can maintain an index, but needs reliable committed changes, replay, ordering, idempotency, deletes, and repair. It does not remove lag. Without a suitable index or alternate access path, queries may require scans.
Rebalancing: three distinct strategies
Growth, new machines, and failures require moving load. Rebalancing should preserve reads and writes, balance storage and traffic, and minimize transfer. Avoid hash(key) mod N when N is the node count: changing N remaps most keys.
Fixed partition count
Create more partitions than nodes and move whole partitions to new owners. Keys keep their partitions; stronger nodes can own more. Too few partitions cap useful distribution; too many add overhead. With a fixed count, growing data produces larger partitions and costlier recovery.
flowchart LR
A[Node A: P1 P2 P3] -->|move P3| C[New node: P3 P6]
B[Node B: P4 P5 P6] -->|move P6| C
💡 The old assignment serves traffic during transfer. Copying, catching up concurrent writes, and coordinating cutover keep ownership changes safe. Partition-to-node assignment changes without reassigning each key to a new logical partition.
Dynamic partitioning
Split oversized partitions and merge small neighbors; move a resulting half if needed. Partition count follows data volume. This works for key ranges and hash ranges. An empty database may start with one partition, so pre-splitting avoids an initial single-node bottleneck; useful key-range boundaries require distribution knowledge.
Partitions proportional to nodes
Keep roughly a fixed number per node. A joining node splits selected hash ranges and takes halves. Many ranges smooth random imbalance; better allocation can improve it. Unlike dynamic partitioning, the count follows machines. Adding nodes shrinks average partition size for a fixed dataset.
💡 Pre-splitting and size thresholds do not guarantee relief during a hot-key write flood. Automatic rebalancing can also mistake overload for failure: copying adds load, causes more timeouts, and triggers cascading movement. Throttling or human approval can prevent this feedback loop.
Routing follows ownership changes
Three options exist: contact any node and let it forward; use a partition-aware routing tier; or let an informed client contact the owner directly. The intermediary returns the owner's response. Whichever component routes must learn current assignments.
flowchart LR
A[Client A] --> N[Any node] --> O[Owner]
B[Client B] --> R[Routing tier] --> O
C[Informed client] --> O
In the book's coordination-service design, nodes register with ZooKeeper, which maintains the authoritative partition map. Routers or clients subscribe to changes. Other designs distribute cluster state through gossip or use database-specific configuration services.
flowchart TB
N[Database nodes] -->|register| Z[Coordination service]
Z -->|assignment updates| R[Router or informed client]
R -->|data request| N
💡 Routing may live in a driver, database node, or dedicated service. etcd can support coordination; a Redis cache alone does not supply an authoritative ownership protocol. Generic gateways and connection proxies do not automatically understand sharding. DNS locates an initial endpoint, while the changing partition map directs individual keys.
Parallel queries and the remaining hard problem
MPP databases divide joins, filters, grouping, and aggregation into execution stages that run across partitions. Large scans benefit from parallelism; exchanges between stages add network and coordination cost. This is more sophisticated than a single-key lookup or simple scatter/gather.
My checklist is to examine the partition key, range locality, hot keys, index fan-out, all three rebalancing options, live cutover, and routing metadata together. Mostly independent partitions enable scale. A write spanning them reintroduces the hard question: what if one succeeds and another fails?
These are my personal learning notes from Designing Data-Intensive Applications by Martin Kleppmann.