Distributed Systems · Partitioning

Why Append-Only Keys Concentrate Load on One Shard

Monotonic keys place every insert at the end of the key space, so one partition absorbs the write traffic. The fixes, from bit-reversal to bucket suffixes, and what each costs.

A range-partitioned store with a key of the form timestamp | sequence looks balanced when examined as a whole and is pathologically unbalanced when examined as a stream. Every insert lands at the top of the key order, which is the last partition, which is one machine. The rest of the cluster holds history and receives nothing. Splitting does not help, because the split creates a new last partition and the writes move into it. The load follows the end of the key space, and the end of the key space is always one partition.

The mechanism, in one trace

Consider a table keyed by a 64-bit millisecond timestamp, with boundaries placed at 1 January of each year as Range Partition Boundaries and the Split Key Contract describes. Data accumulates from 2020 onward. In 2020 all writes go to the 2020 partition. When it fills, it splits, and the writes continue in the upper half, which is the later months. When 2021 arrives, the writes move to the 2021 partition, and the 2020 partition becomes cold.

At any instant, exactly one partition is receiving writes, and it is the one holding the largest key. Its size grows until it exceeds the split threshold, at which point it splits into two, and the upper half immediately starts growing again. The partition count rises steadily while the number of partitions doing useful write work stays at one.

Reads are less skewed, because a workload that queries recent data reads the same hot partition and a workload that queries history spreads across the cold ones. A system that writes and then reads recent data therefore has one partition doing all the writing and all the reading, and the other partitions serving only occasional historical queries.

The cost is not just throughput. The hot partition’s storage engine is also doing the most compaction, because it is the one receiving the most new data, and compaction competes with the writes for disk bandwidth. A hot partition under sustained insert load spends a large share of its I/O on rewriting its own recently written data. The amplification is the storage engine’s normal behaviour, concentrated on the one partition that cannot afford it.

Why hashing alone does not fix the ordering requirement

The obvious remedy is to hash the key, which spreads inserts uniformly. It also destroys the reason the key was ordered: range scans by time, which are the dominant query for time-series and log data, become full fan-outs. A design that hashes loses the ability to ask for “the last hour of events” without touching every partition, and that is usually the whole point of the table.

The tension is real and the resolution is not to pick one side. It is to make the physical key differ from the logical key, so that the sort order used for placement is not the sort order used for queries. The three standard constructions each do this differently.

Prefix bucketing

Prepend a small bucket number to the key: bucket | timestamp | id, with the bucket chosen as a hash of something that varies, such as the id, modulo a small constant like 16 or 64. The physical key order is now bucket-major, so writes spread across as many partitions as there are buckets, and within a bucket the order is still by time.

A range query for a time interval becomes one query per bucket, each of which is a range scan within that bucket. With 16 buckets the query fans out 16 ways, which is a bounded and predictable cost, and each of the 16 scans is efficient because the data within a bucket is ordered by time. The query is 16 parallel seeks rather than one seek, and the merge is a 16-way merge of already-sorted streams.

import hashlib

BUCKETS = 16

def physical_key(row_id: str, ts_ms: int) -> bytes:
    b = int.from_bytes(hashlib.blake2b(row_id.encode(), digest_size=4).digest(), 'big') % BUCKETS
    return b.to_bytes(1, 'big') + ts_ms.to_bytes(8, 'big')

# a time-range read must open one scan per bucket, all with the same bounds
def scan_bounds(lo_ms: int, hi_ms: int) -> list[tuple[bytes, bytes]]:
    return [(b.to_bytes(1, 'big') + lo_ms.to_bytes(8, 'big'),
             b.to_bytes(1, 'big') + hi_ms.to_bytes(8, 'big')) for b in range(BUCKETS)]

The bucket count is the tuning knob, and it is a direct trade. More buckets means more write spread and more read fan-out. The count must be fixed before data is written, because changing it re-buckets every row: a key stored under bucket 3 must be rewritten under a different bucket, which is a full-table rewrite. Systems that anticipate growth choose the bucket count for the peak, accepting a read fan-out that is larger than needed early on. A count that is a power of two is common because the modulo becomes a mask.

The weakness is that bucketing multiplies the number of keys a scan must touch without multiplying the partitions. Sixteen buckets do not necessarily map to sixteen partitions; if the bucket prefix is hashed into the partition space, the buckets distribute across partitions, and a query touches as many partitions as the buckets happen to occupy. With more buckets than partitions, several buckets share a partition and the fan-out is over partitions, not buckets, which is cheaper. With fewer buckets than partitions, the writes still concentrate: 16 buckets cannot spread across 100 partitions.

Bit-reversed and interleaved keys

A more elegant construction reverses the bits of the timestamp before using it as the sort key. Reversal turns the most-significant-bit-first order into least-significant-bit-first order, so consecutive timestamps map to keys that are far apart in the reversed space: timestamp 0 and 1 differ in the top bit of the reversed key, and timestamps differing by 2^k differ at bit k from the top. The result is that new keys spread across the whole space while nearby timestamps remain adjacent in groups: a range of 2^k consecutive timestamps maps to a set of 2^k reversed keys that share their low bits and differ only in the top k, so the group occupies a contiguous, aligned block of the reversed key space.

That property is what makes it usable. Bit reversal maps a block of 2^k timestamps aligned to a multiple of 2^k onto a block of 2^k reversed keys that is itself contiguous and aligned, because the block’s low k timestamp bits, which vary freely inside it, become the high k bits of the reversed key while the block’s high bits stay fixed and become the reversed key’s low bits. So a query for a time range that is aligned to a power of two touches one contiguous run of reversed keys. A query for an arbitrary range touches at most twice as many runs as the range spans in aligned blocks, which for a range of width w is O(log w) runs rather than w separate seeks. The alignment requirement is a real constraint: queries must be expressed in aligned intervals, or the run count grows.

Bit reversal is a bijection on the key space, so it loses no information and needs no extra column. Its cost is that the relationship between key and time is no longer readable by eye, which matters for debugging and for any operation that seeks by time. Some systems keep both a reversed key for placement and the raw timestamp as a value for filtering.

A related construction interleaves the digits of the timestamp with a hash of the id, producing a key whose prefix is a coarse time bucket and whose suffix is a hash. This is a Z-order or Morton-style layout: it spreads writes within a time bucket while keeping time as the dominant order. It gives a bounded write spread and a bounded read fan-out, with the fan-out set by the coarse bucket width rather than by an explicit count.

Salted keys with a lookup

The construction that fully decouples write spread from read fan-out is to choose the salt at write time from a small set and record it. The key becomes salt | timestamp, and the salt is chosen per write batch rather than per row, cycling through the available values. A reader who needs a time range must know which salts are in use, which is a small metadata lookup, and then scan each salt’s range.

This differs from prefix bucketing in that the salt is not derived from the row, so a single logical row’s placement can be chosen for load reasons rather than determinism. The cost is that the salt must be stored with the row or derivable, and that a point lookup by logical key now requires either storing the salt in an index or trying every salt. For a table whose primary access is by time range and whose rows are not looked up individually, that is acceptable; for a key-value table it is not.

A variant used for counters and other hot single keys splits one logical key into a fixed number of physical keys with a suffix 0..k-1, and reads sum across them. That converts one hot key into k warm keys, at the cost of k reads per logical read and a write that must pick a suffix. It is the only technique that addresses a single hot key, which prefix bucketing cannot: prefix bucketing spreads different keys, and a single key remains single.

What the hot partition does to everything else

The skew propagates beyond the write path, and each propagation is a separate failure.

Compaction on the hot partition runs continuously, and its background I/O competes with foreground writes. The storage engine’s write stall, which normally appears only during extreme bursts, becomes a steady state on the one partition. The stall manifests as elevated write latency on that partition, which the coordinator sees as one slow node and the whole cluster sees as a slow write path, because every write goes there.

The hot partition’s leader, if the partition is replicated, is the busiest node for both reads and writes, and its replicas are also busy because they receive the same write stream. A three-replica partition with a hot leader means three machines doing the work of one partition’s load, and none of them can take over if the leader fails without inheriting the same problem. The failure domain is not degraded by the skew; it is amplified by it, because the hot partition’s replicas are all hot.

Splitting the hot partition does not relieve it. The split creates a new upper partition that takes the newest keys, so the write traffic moves to the new partition, on the same machine, and the machine’s total load is unchanged. The number of partitions rises, the metadata service does more work, and the write throughput stays at one machine’s capacity. This is the specific reason that automatic splitting is not a rebalancing strategy: it addresses partition size, not partition load.

Choosing the construction

The decision reduces to three questions with concrete answers.

If the workload is append-only writes with time-range reads, prefix bucketing or bit reversal is appropriate, and the bucket count or alignment width sets both the write spread and the read fan-out. A fan-out of 16 to 64 is usually acceptable; a fan-out of 1000 is not, because the tail latency of a 1000-way scatter dominates.

If the workload has a small number of very hot keys, no key-prefix scheme helps, and the answer is suffix splitting or caching in front of the store. The suffix count should be chosen so that each physical key’s load is below the partition’s capacity with headroom, and the read path must aggregate.

If the workload is genuinely uniform in key space and access, hashing is correct and the ordering requirement does not exist. The mistake to avoid is choosing an ordered key for a workload that does not need order, which imports the skew for no benefit.

In every case the placement decision should be made before the data is written, because the alternatives all require a rewrite. A table that starts with a monotonic key and later decides to bucket must re-key every row, which is a full migration at the scale of the whole table. The migration machinery in Shard Migration Cutover and the Dual-Write Window can perform the move, but it moves the entire data set rather than a fraction of it, and the operation is measured in hours or days.

References

  1. Bigtable: A Distributed Storage System for Structured Data — Chang et al., OSDI 2006
  2. HBase: The Definitive Guide — Lars George, O'Reilly Media, 2011
  3. Dynamo: Amazon's Highly Available Key-value Store — DeCandia et al., SOSP 2007