Sharding: Picking a Key You Won't Regret in Two Years

Problem Context

Vertical scaling has a ceiling — and a punishing price curve as you approach it. At some point your single Postgres or SQL Server box runs out of CPU, RAM, IOPS, or all three, and the answer stops being "a bigger box" and starts being "more boxes, each owning a slice of the data". That's sharding: horizontal partitioning of a logical dataset across N physical stores so that any single shard handles 1/N of the work.

Sharding solves a real problem (write throughput, dataset size, blast radius) and creates several new ones (cross-shard joins, hot shards, rebalancing, distributed transactions). This guide is when to reach for it, how to pick a shard key you won't regret, and the operational realities that don't make it into the design doc.

Concept Explanation

Three sharding strategies cover almost every system:

• Hash-based — shard = hash(key) % N. Even distribution, but adding a shard reshuffles everything.Consistent hashing fixes the reshuffle.
• Range-based — slices by key range (e.g., user IDs 0-1M, 1M-2M). Trivial range scans; suffers from hot ranges (latest data, alphabetical names).
• Directory-based — a lookup service maps key → shard. Maximum flexibility (you can rebalance arbitrarily), at the cost of a lookup hop and a single-point dependency.

flowchart LR
    APP["App"] --> R["Shard Router<br/>(consistent hash ring)"]
    R -->|tenant_id hash A-F| S1["Shard 1<br/>tenants 0-25%"]
    R -->|tenant_id hash G-M| S2["Shard 2<br/>tenants 25-50%"]
    R -->|tenant_id hash N-S| S3["Shard 3<br/>tenants 50-75%"]
    R -->|tenant_id hash T-Z| S4["Shard 4<br/>tenants 75-100%"]

    S1 -.->|cross-shard query| AGG["Scatter-Gather<br/>(slow path)"]
    S2 -.-> AGG
    S3 -.-> AGG
    S4 -.-> AGG

    style R fill:#dc2626,color:#fff,stroke:#b91c1c
    style AGG fill:#f59e0b,color:#000,stroke:#d97706

Implementation

Step 1: Pick a shard key — the most consequential decision

A good shard key:
  ✅ High cardinality           (millions of distinct values)
  ✅ Even access distribution   (no celebrity hot spots)
  ✅ Present on every query     (so the router can target one shard)
  ✅ Immutable                  (changing the key = moving the row)

For multi-tenant SaaS:   tenant_id            ← almost always right
For social graph:        user_id              ← but power users are hot
For time-series:         (device_id, hour)    ← composite to spread hot devices
For e-commerce:          customer_id          ← orders co-locate naturally

Step 2: Use consistent hashing — survive adding shards

// Naive: shard = hash(key) % N → adding a shard moves ~(N-1)/N keys.
// Consistent hash with virtual nodes → adding a shard moves ~1/N keys.
public class ConsistentHashRing<TNode>
{
    private readonly SortedDictionary<uint, TNode> _ring = new();
    private const int VNodes = 200;

    public void AddNode(TNode node)
    {
        for (int i = 0; i < VNodes; i++)
            _ring[Hash($"{node}:{i}")] = node;
    }

    public TNode GetNode(string key)
    {
        var h = Hash(key);
        // first vnode at-or-after h, wrapping around
        var match = _ring.FirstOrDefault(kv => kv.Key >= h);
        return match.Value is null ? _ring.First().Value : match.Value;
    }

    private static uint Hash(string s) => XxHash32.HashToUInt32(System.Text.Encoding.UTF8.GetBytes(s));
}

Step 3: Route in the data access layer

public class ShardedRepository(ConsistentHashRing<string> ring, IConnectionFactory factory)
{
    public async Task<Order?> GetOrderAsync(Guid tenantId, long orderId)
    {
        var shard = ring.GetNode(tenantId.ToString());        // route by tenant
        await using var conn = factory.Open(shard);
        return await conn.QuerySingleOrDefaultAsync<Order>(
            "SELECT * FROM orders WHERE tenant_id=@tenantId AND id=@orderId",
            new { tenantId, orderId });
    }

    // Cross-shard query: scatter-gather (avoid in hot paths)
    public async Task<List<Order>> SearchAcrossTenantsAsync(string sku)
    {
        var tasks = ring.AllNodes().Select(async node =>
        {
            await using var c = factory.Open(node);
            return await c.QueryAsync<Order>(
                "SELECT * FROM orders WHERE sku=@sku LIMIT 100", new { sku });
        });
        var results = await Task.WhenAll(tasks);
        return results.SelectMany(r => r).ToList();
    }
}

Step 4: Co-locate related data

If orders and order_items live on different shards keyed differently, every order detail query is a cross-shard join. Shard child tables by the parent's shard key (composite primary key(tenant_id, order_id, item_id)). One round trip, one shard.

Step 5: Resharding — the dual-write window

Phase 1 — Add new shard, run dual-writes:
  app writes → old shard (canonical) AND new shard

Phase 2 — Backfill historical data:
  background job copies tenant ranges that should move

Phase 3 — Switch reads:
  for migrated tenants, read from new shard; verify checksums

Phase 4 — Cut writes:
  flip directory entry; new shard is canonical; old shard read-only

Phase 5 — Drop old data after retention window

Step 6: Use a managed sharded store when you can

Azure Cosmos DB (NoSQL/Mongo/Cassandra APIs):
  • Built-in partition key → automatic shard routing
  • Auto-scale RU/s; cross-partition query supported but expensive
  • Pick a key with high cardinality and even RU consumption

Azure SQL Hyperscale + Elastic Database Tools:
  • Shard map manager handles routing
  • Per-shard backups; online sharding moves

Citus (PostgreSQL extension):
  • CREATE TABLE ... PARTITION BY HASH; distributed JOINs work transparently
  • Ideal for SaaS multi-tenant on PG

Pitfalls

1. Sharding before you need it.A modern Postgres or SQL Server on a 64-vCPU box handles tens of thousands of TPS. If you're sharding at <5k TPS, you're likely solving an indexing or query plan problem in disguise.

2. The wrong shard key. Choosing a low-cardinality or skewed key (e.g., country_code for a US-heavy app) gives you one shard with 80% of the load and three shards idle. Cardinality + evenness, always.

3. Cross-shard joins as a daily query. Scatter-gather queries are 10-100x slower than single-shard. If your top-10 queries cross shards, your shard key is wrong.

4. Distributed transactions. Two-phase commit across shards locks rows on N machines and is a cascading-failure magnet. Refactor to single-shard transactions; use the saga pattern for multi-shard workflows.

5. Resharding under load with no plan. Adding a shard naively rebalances every key — your latency triples for hours. Consistent hashing + dual-write rollouts make this routine.

6. Treating shards as a backup. Each shard needs its own backup, PITR, and DR plan. Restoring shard 7 from yesterday while shards 1-6 stay current produces inconsistent state.

Practical Takeaways

• Don't shard until indexing, caching, and read replicas are exhausted. It's a one-way door.
• Pick the shard key on day one and never change it. Tenant ID is the right answer for most SaaS.
• Use consistent hashing with virtual nodes. Avoid hash % N.
• Co-locate related rows by sharing the parent's shard key. One query, one shard.
• Avoid distributed transactions. Use sagas for cross-shard workflows.
• Plan resharding as a dual-write phased rollout, not a maintenance window.
• Prefer managed sharded stores (Cosmos DB, SQL Hyperscale, Citus) over hand-rolled routing if you can.