Partitioning, Sharding, Skew, and Rebalancing

The key review has three proposals, and each is correct about one access path.

Ledgerline stores time-ordered entries for thousands of tenants. Most tenants are small. One tenant produces 38% of current write service demand during settlement and another issues long historical reads. The critical operations are an exact lookup by entry ID, a tenant-scoped time-range read, and an append whose ordering boundary is explicit.

The proposals are:

• partition by tenant_id, so a tenant range read stays local;
• partition by hash(entry_id), so writes spread evenly; or
• partition by event_hour, so recent scans touch few ranges.

Tenant-only makes the busiest customer one unsplittable partition. Entity hashing turns every tenant time-range read into broad scatter. Timestamp-only sends every tenant’s current writes into the newest range and weakens tenant isolation. The review cannot select a key until it ranks capacity, concurrency, locality, isolation, fault containment, movement, and future growth for the actual operation mix.

Ledgerline’s resulting design uses a directory-backed hierarchy. Ordinary tenants receive coarse tenant/time ranges. A hot tenant receives smaller time buckets with a deterministic subshard derived from entry identity; its range read fans out to four known subshards rather than the whole fleet. The directory carries placement and authority epoch, so a range can split or move without changing the public identity of every record.

This is not the one universal partition scheme. It is a method: partition on the dimension that best balances dominant service demand while preserving locality for critical operations, and add indirection when movement is inevitable.

Partitioning is demand placement with ownership

Partitioning divides data or work into independently placeable units. Sharding commonly means distributing those units across nodes or services; usage varies, so define the term in the design. The partition function is not merely a storage-layout choice. It determines which operations are local, which queues share fate, where coordination crosses boundaries, how failures spread, and what must move during growth.

State the goals before the key:

The unit of ownership may be a key range, hash bucket, tenant, geographic cell, semantic aggregate, or virtual partition. It must be large enough to administer and small enough to place, split, recover, and isolate. Millions of tiny partitions burden metadata, routing, file handles, compaction, and control loops. A few giant partitions make hot spots and movement indivisible.

Correctness constraints come before balance. If all mutations of an account require one serial order, distributing them across independent owners does not remove that invariant. It moves coordination into a transaction or merge protocol. If operations within a hot tenant are independent by entity or time bucket, refine the key along that dimension. If one object is intrinsically hot and ordered, replicate reads, reduce per-operation work, batch compatible updates, or admit less; hashing the same key still selects one owner.

Compare strategies by their slow and failure paths

Range partitioning

Range partitioning assigns contiguous ordered keys, such as [tenant, timestamp], to owners. It supports ordered scans, prefix queries, and adjacent splits. It preserves locality when the dominant read follows the key order.

Monotonic keys can create a moving hot range: every new timestamp lands at the high end. Uneven tenant sizes produce uneven ranges. Split points need demand-aware selection, not equal keyspace width. A range move carries its current write stream, recent cache state, secondary indexes, and change history—not only its snapshot bytes.

Hash partitioning

A hash spreads sufficiently varied keys across buckets and usually improves write distribution. It supports exact lookup when the full partition key is known. It destroys natural order, so range and prefix queries scatter unless another index restores locality. The hash does not remove skew when many requests target the same key, nor does it guarantee balanced service demand when key costs differ.

Changing a simple modulo from (N) to (N+1) can remap most keys. Consistent hashing or a stable bucket layer reduces unnecessary movement by mapping many logical buckets onto fewer physical owners. Its balance and remapping properties depend on token placement, virtual-node count, weights, and workload. It is a placement mechanism, not proof that an application’s access pattern is good.

Directory partitioning

A directory maps a stable logical partition identifier to current owner, key bounds, epoch, and state. It can make placement nonuniform: large tenants get more subpartitions; regulated tenants remain in permitted regions; hot ranges move independently.

Indirection adds a metadata dependency, routing cache, invalidation protocol, and stale-route behavior. The directory must be partitioned and replicated without becoming one global lock. A stale client should be redirected by an authoritative owner with a newer epoch; it must not let two owners accept conflicting writes. Cache entries need bounded staleness or push invalidation, and moves need versioned atomic publication.

Geographic and semantic partitioning

Geographic partitioning places state near users or inside residency boundaries. It lowers local distance and can contain regional failure, but global users and cross-region operations pay coordination or stale-read costs. Geography is often a hard eligibility constraint followed by range or hash placement within the eligible region.

Semantic partitioning follows domain boundaries: merchant, account, room, project, model, or graph community. It can preserve invariants and co-access locality better than opaque hashing. Domain skew and organizational change make its partitions uneven. A semantic boundary that feels stable today can become the next acquisition, mega-tenant, or viral object.

Use combinations deliberately:

A good key serves more than one access dimension

A partition key should be evaluated against the operation matrix, not judged by cardinality alone. For every significant operation, record its key fields, frequency, payload, service demand, result size, locality requirement, ordering or transaction boundary, deadline, and growth.

High cardinality helps create choices; it does not ensure even popularity or cost. A million keys with one key receiving half the demand remains skewed. A uniform request count can hide a key whose queries scan a thousand times more rows. Measure per-key and per-partition CPU time, device bytes, lock/transaction hold time, queue occupancy, storage bytes, network, cache churn, and downstream fan-out.

Multi-dimensional access creates an unavoidable choice. Ledgerline needs both entry lookup and tenant time-range reads. The directory-backed composite scheme works as follows:

logical key: tenant_id / time_bucket / subshard / entry_id

ordinary tenant:
  directory(tenant, week) -> one range owner

hot tenant:
  directory(tenant, hour, hash(entry_id) mod 4) -> four range owners

point lookup:
  entry locator or encoded bucket -> one owner

tenant time-range read:
  directory enumerates overlapping buckets -> bounded parallel read + ordered merge

The subshard count is metadata, not a global constant. New writes use the current epoch. Historical buckets stay at their old width unless movement benefits justify rewriting them. This avoids a full historical reshard whenever a tenant grows. It also means readers must understand multiple bucket generations and merge them correctly.

The fixture compares four candidates:

These modeled values assert no database capacity. They make the decision legible: a four-way read is accepted because it preserves the critical range operation within a bounded set while bringing the hottest owner below the teaching threshold. If production evidence shows merge CPU or tail amplification dominates, use fewer subshards, precompute summaries, change the index, or revisit the product query.

Partition keys are security and privacy boundaries too

Never trust a client-supplied tenant key without authorization. Route using server-validated identity. Include tenant scope in caches, secondary indexes, logs, encryption context, backup, deletion, and movement records. A directory entry that redirects tenant A into tenant B’s range is a data exposure, not a performance bug.

Geographic and legal eligibility may constrain owners before load balancing. Encryption keys, retention, deletion, and audit evidence must move or remain accessible through the entire lifecycle. A partition snapshot copied for rebalancing is another data replica with access control and cleanup obligations.

Detect skew in demand, not inventory

The fixture’s six partitions hold between 9.8 and 10.2 million items. Their item spread is about 4% of the mean—an attractive inventory dashboard. Service demand ranges from 520 to 4,680 units/s. The maximum is more than 3.7 times fleet mean.

Inventory balance can coexist with demand imbalance because popularity, operation mix, item size, scan depth, cache miss rate, write amplification, and compaction differ. Build heat maps over the dimensions that explain demand:

• partition by time, tenant, operation, and outcome;
• key/range popularity and concentration;
• request count and normalized CPU/I/O/network demand;
• queue age, concurrency, service time, and useful completion;
• bytes/items plus index, version, and tombstone density;
• cache hit/miss and remote-access rate;
• background compaction, replication, backup, and movement work; and
• failed/recovering state, not only normal traffic.

Use distributions across partitions. Track maximum-to-median, high percentile, top-k concentration, and share above a safe operating bound. A fleet average cannot identify the owner whose queue is failing. A coefficient or fairness index can summarize, but always retain the ranked raw demand and the reason one partition differs.

Distinguish hot key, hot partition, and hot node. One hot key may be intrinsically serialized. A partition may contain many moderately hot keys that can be split. A node may host several ordinary partitions whose combined resource vector exceeds capacity. The remedy and movement unit differ.

Skew also changes over time. A tenant/time heat map can show a vertical stripe for a persistently large tenant, a horizontal band for the current hour across tenants, and their intersection as the critical hot partition. Design the bucket hierarchy to address the causal dimension. Randomly moving the intersection to another node only moves the overload.

Virtual partitions and indirection buy movement granularity

Virtual partitions separate the logical partition count from the physical owner count. A new node can receive some virtual units instead of changing the hash of every record. Heterogeneous nodes can receive weighted counts. Failure recovery can spread a lost owner’s units across several destinations.

The benefits have costs:

• routing and ownership metadata grows;
• each unit may carry files, indexes, compaction, cache, and metrics overhead;
• too many simultaneous moves saturate shared network and storage;
• random tokens can fragment ordered scans;
• a virtual unit can still be too large or hot to move safely; and
• token count balances opportunity, not service demand by itself.

Choose initial units from projected data and demand, smallest practical movement size, metadata overhead, rebuild/recovery time, and maximum concurrent moves. Preserve spare units or split capability so future growth has somewhere to go. Measure the distribution after applying physical weights and failure-domain constraints.

Consistent hashing is particularly useful when assignment churn should move only a subset of independent keys. Range directories are often better when ordered scans and targeted splits dominate. A system can hash tenants to groups and range-partition time within each tenant group. Do not import a ring merely because the design is distributed.

Secondary access paths move the cost elsewhere

When the primary partition key does not support an operation, choices include scatter-gather, a secondary index, a locator directory, materialized view, search system, or changed product contract. Each creates state and failure behavior.

A global secondary index turns one write into at least a primary update plus index update. Define whether they commit atomically, converge asynchronously, or expose a version boundary. A stale index can point to a moved or deleted record. Rebalancing may require rebuilding index entries or preserving a logical partition ID so physical movement does not rewrite every pointer.

A local secondary index is cheaper to keep consistent with its partition but requires scatter across partitions for queries without the primary key. A global index narrows reads but becomes its own partitioned, skewed system. Popular index values can be hot keys. Unique constraints can require coordination at the index owner.

Measure write amplification, index lag, lookup fan-out, false/missing candidates, repair traffic, storage, and recovery. State which source is authoritative. If an asynchronous index omits a recent row, the API must declare the freshness semantics rather than calling it a timeout.

Co-partition data that is frequently joined or transacted together when the common key and lifecycle are stable. Co-partitioning saves network and coordination on the dominant path but couples movement and growth. A large shared dimension can make one owner enormous. Sometimes shipping a small dimension table, maintaining a bounded materialized view, or accepting a remote join is cheaper than forcing all data into one layout.

Data gravity includes more than stored bytes: warm cache, indexes, logs, checkpoints, replica state, network proximity, hardware specialization, and consumers that assume locality. Affinity is the measurable benefit of keeping those related states or operations together; it must be priced against skew, failure concentration, and movement cost. Moving 512 GiB of cold snapshot may be easier than rebuilding a 40 GiB hot index while serving traffic. Inventory every attached state before calling a partition movable.

Scatter-gather amplifies tails and work

A scatter query sends subrequests to (k) partitions and commonly completes when all required results arrive. Even if individual latencies are identically distributed, the maximum tends to rise with (k). Real branches are not independent: they share network, coordinator CPU, caches, and failure domains.

The cost model includes:

[ W_{query} = \sum_{i=1}^{k}(W_{route,i}+W_{scan,i}+W_{return,i}) + W_{merge} ]

and completion latency is approximately the slowest required branch plus coordinator and merge work. Pagination, top-k, aggregation, and early termination change the formula but do not make fan-out free.

Bound:

• maximum eligible partitions and branch concurrency;
• per-branch remaining deadline and cancellation;
• bytes/rows scanned and returned;
• coordinator memory and merge CPU;
• partial-result correctness and freshness;
• retry ownership and duplicate branches;
• admission price proportional to predicted fan-out; and
• behavior when a partition is moving, failed, or recovering.

For the hot tenant, four known subshards are a bounded compromise. The coordinator sends the same snapshot/version requirement, merges by timestamp and entry ID, stops after the requested range and page limit, and cancels surplus prefetch. A missing shard produces an explicit incomplete or failed result according to the query contract; it is not silently omitted.

Hedging every branch of a 16-way scatter can turn 16 requests into 32. Retrying at both coordinator and branch layers recreates Chapter 25’s multiplication. Keep one logical deadline and retry owner. Chapter 27 will add replica selection; partition fan-out and replica fan-out must be multiplied in the capacity model.

Splits, merges, and moves are production workloads

A split divides one logical unit into smaller ownership units. A merge combines adjacent or compatible units. A move transfers an unchanged unit to another owner. Resharding may change the partition function or keyspace organization. State which operation is occurring; their correctness and cost differ.

Trigger movement from sustained demand, capacity, failure-domain, maintenance, or policy evidence—not one noisy sample. Include hysteresis and residence time so the controller does not move a tenant back and forth. Placement from Chapter 23 chooses destinations; admission from Chapter 24 reserves headroom; Chapter 25 supplies deadline, retry, idempotency, and ambiguity rules for control operations.

The movement budget includes source reads, destination writes, checksums, change capture, dual writes, index work, compaction, cache warm-up, replica work, and user traffic displaced from storage or network. Limit concurrent moves per source, destination, rack, zone, and shared device. Reserve recovery bandwidth separately; routine balancing must yield to a failure rebuild.

Merge cold tiny partitions to control metadata only when their combined future demand and failure blast radius remain acceptable. A merge can be as risky as a split because it changes routing and authority. Keep a rollback or forward-repair path until old metadata and data are safely retired.

Applied plan: split 512 GiB under live traffic

The modeled hot range contains 512 GiB. The source observes 30 MiB/s of live writes. The migration link budget is 96 MiB/s after user-traffic and recovery reserves. Choose a 64 MiB/s snapshot copy, leaving 32 MiB/s. Snapshot copy plus the modeled live stream peaks at 94 MiB/s, inside the budget with only 2 MiB/s modeled margin. That is intentionally tight evidence: a real plan would add measurement uncertainty or lower the copy rate.

At 64 MiB/s, copying 512 GiB takes:

[ 512 \times 1024 / 64 = 8192\ seconds \approx 136.5\ minutes ]

Do not dual-write the entire 136.5-minute snapshot if a change log can capture the live delta. Use five stages:

1. Snapshot. Directory epoch 41 names the old owner as authoritative. Establish a consistent snapshot boundary and durable change position. Create destination partitions in COPYING; clients cannot route writes to them.
2. Copy. Stream snapshot chunks at the bounded rate with content/range checksums and resumable chunk identities. The old owner records post-snapshot changes. Abort or throttle when user SLO, device queue, recovery reserve, or checksum health crosses a guardrail.
3. Catch up and short dual write. Apply the change log in order. When lag fits the cutover window, the old owner remains the only external write authority but synchronously forwards or coordinates writes to both old and new layouts. In the modeled twelve-minute interval, 30 MiB/s creates about 21.1 GiB of duplicate writes. The dual path has one idempotent mutation identity and explicit acknowledgment rule.
4. Validate and cut over. Compare counts by range, hashes/checksums, high-water positions, sampled records, secondary indexes, and invariant queries. Atomically publish epoch 42 with new key bounds and owners. Fence epoch 41 writes: the old owner redirects or forwards them under the new epoch; it cannot continue accepting independent commits. Retain rollback only while authority remains unambiguous.
5. Clean up. Drain stale readers and in-flight operations, verify no old-epoch writes, rebuild/warm required indexes and caches, then tombstone and eventually delete old data under retention and audit rules. Release movement reservations gradually.

“Dual write” here does not mean two independent primaries. One authority coordinates a bounded duplicate path. If either destination cannot meet the acknowledgment contract, stop or remain at the old epoch. A crash between writes is recovered by the mutation identity and change log. The state machine must define restart behavior in every stage.

Authority and rollback ledger

Rollback before cutover means discard or repair the candidate while epoch 41 remains authoritative. After cutover, “roll back” is another forward authority transfer with a newer epoch; reactivating epoch 41 invites split authority. Keep source data long enough for recovery, but do not confuse retained bytes with permission to write.

Rebalancing must remain stable under failure and recovery

An automatic controller observes delayed demand, chooses moves, copies for hours, and only later changes load. By then the hot period may have ended. If it chases current CPU without predicting movement cost and residence time, it can oscillate and spend more capacity moving than serving.

Use slow signals for ownership: sustained normalized demand, projected growth, hard capacity, failure-domain policy, and movement benefit over a minimum residence horizon. Use faster admission and request placement to handle short spikes. A hot partition may be temporarily rate-limited, replicated for reads, or assigned extra compute while a split proceeds.

Failure changes priorities. Losing a node transfers its partitions and rebuild traffic. Do not simultaneously rebalance healthy units merely because destination utilization changed. Quarantine failed units, reserve network/storage for durability recovery, and recalculate capacity after the failure domain—not before. Recovery includes cache coldness, compaction, index build, replay, and client retries.

Test directory unavailability, stale routes, source crash, destination crash, checksum mismatch, change-log gap, cutover timeout, delayed old client, duplicated control message, region partition, and insufficient destination headroom. Every split command needs an idempotent operation identity. Every state transition needs an owner and monotonic epoch.

Partition-key review worksheet

SYSTEM BOUNDARY
Authoritative state and partition/ownership unit:
Physical resources and failure domains:
Normal, peak, failed, recovering, and growth envelopes:

OPERATION MATRIX
Operation / frequency / payload / normalized demand:
Point, prefix, range, join, aggregate, and mutation paths:
Ordering, transaction, freshness, residency, and isolation invariants:
Deadline, fan-out ceiling, result/merge bound:

CANDIDATE KEY
Fields, cardinality, distribution, time evolution:
Range / hash / directory / geography / semantic composition:
Hot-key and mega-tenant behavior:
Local operations; scattered operations; secondary access paths:

BALANCE AND BLAST RADIUS
Per-partition CPU, I/O, bytes, queue, service time, and goodput:
Tenant/class concentration and failure impact:
Virtual-unit size/count, metadata overhead, split floor:
Behavior after node, rack, zone, and directory loss:

MOVEMENT
Split/merge/move triggers, hysteresis, and residence time:
Snapshot bytes, live-write rate, copy/index/cache/recovery demand:
Change capture, idempotent mutation identity, authority epochs:
Validation, fencing, cutover, abort, forward recovery, cleanup:

EVIDENCE
Representative trace or modeled fixture and transfer limits:
Skew, hot key, scatter, failure, migration, and recovery tests:
Rollout cohort, SLO guardrails, pause/abort owner:

Field questions

• Does the balance unit follow the scarce resource or only item count?
• Which critical operations remain local, and what is the maximum scatter?
• Can one tenant, key, time frontier, or semantic aggregate defeat the key?
• Are hard residency and correctness constraints outside the balancing score?
• Who routes a stale client, and which epoch fences the old owner?
• What secondary-index write, lag, and recovery work did the design add?
• How much source, destination, network, cache, and compaction work does movement consume?
• Can a routine move steal bandwidth from failure recovery?
• Is rollback truly safe before cutover, and is post-cutover recovery a newer epoch?
• What evidence would force a key redesign instead of another rebalance?

Design drill: challenge the key, then fail the move

Take the Ledgerline operation matrix and score tenant-only, entity hash, time range, and directory tenant/time/subshard candidates. Weight the score by modeled service demand rather than counting operations equally. Preserve exact lookup, bound the dominant tenant-range read, state the ordering unit, and reject any plan that assumes one intrinsically hot ordered key can be hashed into independence.

For the selected scheme, draft a 512 GiB split using the fixture. Show 8,192 seconds of snapshot copy at 64 MiB/s, 94 MiB/s peak migration traffic when 30 MiB/s of live changes coexist, and about 21.1 GiB duplicated during the twelve-minute dual path. Replace the 2 MiB/s residual margin with a defensible production uncertainty reserve or lower the copy rate.

Now crash the source during copy, during catch-up, immediately before epoch publication, and immediately after it. The answer must identify authority in each case, prove that mutation replay is idempotent, and specify whether the action is resume, discard candidate, remain at epoch 41, or continue forward from epoch 42. “Retry the migration” is insufficient unless the control operation and chunks have stable identities.

Durable decision rules

1. Partition by the dimension that balances dominant service demand while preserving locality for critical operations and correctness boundaries.
2. Treat cardinality as opportunity, not evidence of balance; measure per-partition resource demand, queues, tails, and useful completion.
3. Add virtual partitions or a directory when heterogeneous growth and future movement justify their metadata, routing, and stale-owner complexity.
4. Bound every cross-partition query by branch count, deadline, bytes, merge resources, partial-result semantics, and retry ownership.
5. Price secondary indexes and co-partitioning through write, consistency, recovery, privacy, and movement paths—not only read latency.
6. Run split, merge, move, and reshard as admitted production workloads with source, destination, network, cache, index, and recovery budgets.
7. Transfer write authority through a monotonic epoch and fencing protocol; retained old data never implies retained independent authority.

Evidence and transfer limits

• Dynamo: Amazon’s Highly Available Key-value Store is a primary description of one consistent-hashing, virtual-node, replication, and failure design. Its assumptions do not select a partition key or consistency model for Ledgerline.
• Spanner schema design best practices documents current range-oriented key locality and hotspot cautions for Spanner. Product behavior and recommendations are not universal database laws.
• Apache Cassandra architecture: Dynamo documents Cassandra’s current token-ring and partitioning concepts. Token ownership does not guarantee balanced application service demand.
• Apache Kafka basic operations documents current partition reassignment and migration-throttling mechanisms. Kafka partitions and leaders illustrate one log system; they do not define database resharding semantics.
• CockroachDB load-based splitting documents one current product mechanism that samples access load to decide whether a range can be split. Its thresholds, range model, and interaction with rebalancing require version- and environment-specific validation.
• All inventory, demand, candidate-key, copy, link-budget, dual-write, and epoch values are deterministic modeled evidence in examples/performance-engineering-system-design-handbook/part-03/partitioning-rebalancing/. The fixture has no database, router, directory, persistent log, checksums, faults, replicas, queues, or production measurements.

Partitioning decides which owner holds state and which operations cross ownership boundaries. It does not make those owners durable or available. The next chapter adds replicas and quorum paths, asking when an acknowledgment is safe, what a read may observe, and how recovery traffic changes the partition’s latency and capacity budget.