Partitioning & NewSQL

Three years after launch, BoxOffice sells in 30 countries. The orders and tickets tables grow by hundreds of millions of rows a year, the primary's disk is measured in terabytes, and the write rate during big on-sales is close to what one machine can absorb. Replicas no longer help: they scale reads, and every replica must still apply every write. The next rung of the scaling ladder is to split the data itself across machines.

Partitioning — also called sharding — divides a dataset so that each record belongs to exactly one partition (shard), and different partitions live on different nodes. Each node handles only its share of the reads, writes and storage. The words vary by product — shard in MongoDB and Elasticsearch, region in HBase, tablet in Bigtable, range in CockroachDB, vnode in Cassandra — but the questions are always the same: by what key, how to spread it evenly, how to find it, and how to move it when nodes change¹.

flowchart LR
    R["router<br/>partition = f(key)"] -->|"customer 7"| S1[("shard 1<br/>+ replicas")]
    R -->|"customer 1042"| S2[("shard 2<br/>+ replicas")]
    R -->|"customer 98311"| S3[("shard 3<br/>+ replicas")]

Partitioning is usually combined with replication: each partition has its own leader and followers, so a node can be leader for some partitions and follower for others.

Partitioning by key range

Assign each partition a continuous range of keys, like the volumes of an encyclopedia: A–F, G–M, N–S, T–Z. Within a partition, keys are kept sorted.

• Range queries are cheap: "orders from 2027-03-01 to 2027-03-07" reads one or two partitions.
• Boundaries adapt to the data: when a range grows too large, it is split in two (Bigtable tablets, HBase regions, CockroachDB and TiDB ranges do this automatically).
• Danger — sequential keys. If the key is a timestamp or an auto-increment id, all new writes go to the last range. Today's clickstream lands on one node while the others idle. Fix it by prefixing the key with something else: (event_id, timestamp) instead of timestamp.

Partitioning by hash

Apply a hash function to the key and use the hash to choose the partition. A good hash spreads even sequential keys uniformly.

• Even load for writes and point reads.
• Range queries scatter: adjacent keys hash to unrelated partitions, so "orders this week" must ask every partition.
• Compound keys recover some locality: Cassandra hashes only the partition key and keeps rows sorted by the clustering columns inside it — PRIMARY KEY ((event_id), scanned_at) gives an even spread across events and cheap time ranges within one event.

The naive implementation, partition = hash(key) mod N, has a fatal flaw: when N changes, almost every key changes partition.

Consistent hashing

Consistent hashing² places both nodes and keys on the same circular hash space (a ring, e.g. 0 to 2^64 − 1). Each key belongs to the first node found walking clockwise from the key's position.

When a node joins, it takes over only the keys between itself and its predecessor; when a node leaves, only its keys move to its successor. On average 1/N of the keys move, the minimum possible — compared with almost all of them under mod N.

With only one position per node, the arcs are uneven and one node can own far more than its share. Virtual nodes (vnodes) fix this: each physical node takes many positions on the ring (100–256 is common), so its total share averages out, and a node with twice the capacity can simply take twice the vnodes.

The simulation below hashes 20,000 keys with MD5. On the left, the percentage of keys that change node when the cluster grows from N − 1 to N nodes; on the right, how unbalanced 10 nodes are (busiest node ÷ average) as the number of virtual nodes grows:

Going from 9 to 10 nodes with mod N moves about 90% of the keys — in a distributed cache, that is the same as flushing the whole cache onto the database at once. With consistent hashing, about 10% move. And with a single position per node, the busiest node carries several times its fair share; a few hundred vnodes bring it close to 1.

Consistent hashing is used by Dynamo and DynamoDB, Cassandra and ScyllaDB, Riak, client-side Memcached sharding (the ketama algorithm), Nginx's hash … consistent upstreams and many load balancers. Two relatives are worth recognizing: rendezvous hashing (highest random weight: each key picks the node with the highest hash(key, node))³ and jump consistent hash (no ring, near-perfect balance, but nodes can only be added or removed at the end)⁴.

Fixed number of partitions

A simpler alternative many systems use: create many more partitions than nodes up front and assign whole partitions to nodes. Adding a node means moving some whole partitions to it; the key → partition mapping never changes.

• Redis Cluster: 16,384 hash slots, slot = CRC16(key) mod 16384; slots are moved between nodes (see Redis).
• Elasticsearch / OpenSearch: the number of primary shards is fixed when the index is created.
• Kafka: a topic's partitions (see Kafka).

The catch: choose the count well. Too few partitions limits how far you can scale; too many adds per-partition overhead.

Choosing the partition key

The partition key is the most consequential and the least reversible decision of a sharded design. A good key has:

1. High cardinality — many distinct values, so the data can be cut into many pieces.
2. Even distribution of both data and traffic across values.
3. Query locality — the most frequent queries and every transaction touch one partition.

Hot partitions

Even with a perfect hash, traffic is rarely uniform. A hot partition (or hot key) receives far more load than the others: the celebrity whose posts everyone reads, the product on a Black Friday deal, the stadium show at 10:00. Hashing cannot spread load that targets one key.

Try it. The sharding scenario spreads a NoSQL table over several partitions. Run it with uniform traffic and note the per-partition load; then make the key distribution skewed (one hot key) and watch a single partition saturate and its latency climb while the others are idle — adding partitions does not help.

Mitigations, from the least to the most invasive:

Secondary indexes on partitioned data

Partitioning by customer_id makes "orders of customer 7" fast. What about "orders of event 42"? A secondary index is needed, and on partitioned data it comes in two flavours:

flowchart TB
    subgraph local ["Local (document-partitioned) index"]
        Q1["query: event_id = 42"] --> P1["shard 1<br/>+ its own index"]
        Q1 --> P2["shard 2<br/>+ its own index"]
        Q1 --> P3["shard 3<br/>+ its own index"]
    end
    subgraph global ["Global (term-partitioned) index"]
        Q2["query: event_id = 42"] --> IX["index partition<br/>for event_id 42"]
        IX --> D1["shard 1"]
        IX --> D3["shard 3"]
    end

Scatter/gather is more dangerous than it looks: a query to 20 shards is as slow as the slowest of the 20. The chance that all 20 answer within their own p99 is 0.99²⁰ ≈ 82 %, so a shard's p99 is only about the p82 of the query — 18 % of queries are slower. For the query p99 to stay under a target, each shard must meet it at its p99.95 (0.99^(1/20) ≈ 0.9995) (see Percentiles).

Rebalancing and resharding

Nodes are added, removed and fail; data grows unevenly. Rebalancing moves partitions so that load stays even, ideally while the database keeps serving traffic and moving only the data that must move.

• Prefer schemes where the key → partition mapping is stable (fixed partitions, consistent hashing, range splits), so rebalancing moves partitions, not individual keys.
• Rebalancing competes with live traffic for network and disk; throttle it, and avoid fully automatic rebalancing triggered by a node that is merely slow (it can cascade).
• Changing the partition key (resharding) is a migration project. The common recipe is online:

flowchart LR
    A["1. create new shards<br/>with the new key"] --> B["2. dual-write:<br/>old + new"]
    B --> C["3. backfill history<br/>(copy + CDC catch-up)"]
    C --> D["4. verify:<br/>compare reads"]
    D --> E["5. switch reads<br/>to new shards"]
    E --> F["6. stop old writes,<br/>delete old shards"]

Sharding layers automate much of this for existing relational databases: Vitess for MySQL (built at YouTube) and Citus for PostgreSQL handle routing, resharding and distributed queries.

Request routing

Some component must know which node holds which partition:

The map itself is small, critical metadata — kept in a consensus store such as etcd or ZooKeeper, or gossiped between nodes (see Consensus).

Cross-partition queries and transactions

Partitioning works beautifully until an operation needs several partitions at once:

• Joins and aggregates across shards become scatter/gather, done by a coordinator — slower, and limited by the slowest shard.
• Transactions across shards need a coordination protocol — two-phase commit or consensus-based commit — with extra round trips and failure modes (see Transactions).
• Unique constraints on non-key columns (unique e-mail when sharding by customer_id) need a global index or a separate lookup table.

So the design goal is: choose keys so that the hot path is single-partition, and send the rest to systems built for it (a warehouse for analytics, a search index for cross-cutting lookups).

NewSQL: distributed SQL databases

What if you want partitioning and SQL transactions across partitions, without writing a sharding layer? NewSQL or distributed SQL databases do this⁵:

• Google Spanner⁶ splits tables into ranges, replicates each range with Paxos across zones or regions, and runs distributed transactions with two-phase commit over those groups. Its TrueTime API exposes clock uncertainty (a few milliseconds, bounded by GPS and atomic clocks); a commit waits out that uncertainty, which gives external consistency (linearizable transactions) across the world.
• CockroachDB⁷ (PostgreSQL wire protocol) splits data into ranges of a few hundred megabytes, replicates each with Raft, and uses hybrid logical clocks instead of special hardware; serializable isolation is the default.
• YugabyteDB (PostgreSQL-compatible) and TiDB (MySQL-compatible) follow similar designs; cloud providers offer managed and serverless distributed SQL as well.

sequenceDiagram
    participant C as Client
    participant G as Gateway node
    participant R1 as Range 1 leader (orders)
    participant R2 as Range 2 leader (tickets)
    C->>G: BEGIN, INSERT order, INSERT tickets, COMMIT
    G->>R1: write intent (replicated by Raft)
    G->>R2: write intent (replicated by Raft)
    R1-->>G: ok (majority acked)
    R2-->>G: ok (majority acked)
    G->>G: commit record (atomic)
    G-->>C: COMMIT ok

Distributed SQL is not free: every write is replicated by consensus, so latency is higher than on a single node, and a transaction that spans regions pays inter-region round trips. It is the right answer when you need both SQL transactions and more write capacity or geographic resilience than one primary can give — not as a default.

How it connects

flowchart LR
    RANGE["by key range"] --> KEY["choosing the partition key<br/>cardinality, distribution, locality"]
    HASH["by hash"] --> KEY
    CH["consistent hashing<br/>virtual nodes"] --> HASH
    KEY --> HOT["hot partitions<br/>the on-sale event"]
    HOT --> MIT["cache, finer key, salting,<br/>queue → Caching, Messaging"]
    KEY --> SEC["secondary indexes<br/>local vs global"]
    KEY --> RESH["rebalancing,<br/>request routing"]
    KEY --> XP["cross-partition queries<br/>reports → warehouse"]
    XP --> NEW["NewSQL: Spanner, CockroachDB<br/>consensus per range → Distributed"]
    CH -.-> LB["the same ring in load balancers<br/>→ Edge"]

Palette components: nosql (Cassandra, DynamoDB, MongoDB partition by key) · sql (sharded PostgreSQL, or distributed SQL) · cache (Redis Cluster hash slots; serves the hot event page instead of the shard) · search (Elasticsearch shards) · kafka (partitions by key) · queue (the waiting room in front of a hot partition) · lb (consistent hashing for affinity) · dwh (where scatter-gather reports belong). Scenarios: sharding, flash-sale. Continues in: Specialized Stores, Distributed — consensus, Edge — load balancing, Caching — Redis Cluster, Case study: distributed cache.