Sharding #

Sharding is one of the most impactful and hardest-to-reverse technical decisions in a database system’s lifecycle. Done at the right time and with the right shard key, it lets a system serve billions of rows and millions of writes per day — something no single database server, however advanced, can do. Done too early or with the wrong shard key, it creates complexity that slows a team down for years, turns debugging into a nightmare, and makes every schema change feel like open-heart surgery.

Sharding isn’t the first step in database scalability — it’s the last step, taken after all the simpler techniques have been optimized and are no longer sufficient. Understanding sharding correctly means understanding not just how it works, but also every trade-off that comes with it and every problem it creates while solving another.

Sharding vs Partitioning: A Fundamental Difference #

These two terms are often used interchangeably even though they refer to different concepts at different levels. Understanding the difference matters before deciding which approach is needed.

Partitioning — works within one database instance:

flowchart TD
    subgraph DB["Database Server (single)"]
        direction LR
        p1["orders_2025_01"]
        p2["orders_2025_02"]
        p3["orders_2025_03"]
    end

→ One server, one database process → CPU, RAM, disk still shared → One index and query planner → Foreign keys still usable

Sharding — works at the infrastructure level, across servers:

flowchart LR
    sa["Shard A (server 1)<br>user_id 1–1M"]
    sb["Shard B (server 2)<br>user_id 1M–2M"]
    sc["Shard C (server 3)<br>user_id 2M–3M"]

→ Three separate servers, three database processes → Own CPU, RAM, disk each → No cross-shard foreign keys → Cross-shard queries need application-layer coordination

In practice, both techniques are often combined: data is sharded across several servers, and within each shard, large tables are partitioned by time. This delivers the benefits of both — horizontal load distribution from sharding and query pruning from partitioning.

Why Sharding Is Needed: The Unavoidable Limits #

Every database server has physical limits that no optimization can break through. When a system approaches these limits, sharding becomes the only realistic way out.

The physical limits that force sharding:

  CPU limits:
  → A single server has a limited number of CPUs
  → The more concurrent queries, the more CPU needed
  → Beyond a certain point, adding CPU doesn't linearly increase throughput

  Write throughput limits:
  → A single primary database can only serve N writes per second
  → Writes exceeding this capacity cause queues, lock contention, timeouts
  → Read replicas don't help writes — writes still go to the primary

  Storage limits:
  → A 50TB dataset on one server: backups take days, restores take days
  → Indexes on a 10-billion-row table can be bigger than the server's RAM
  → VACUUM/OPTIMIZE on giant tables blocks other operations

  Operational limits:
  → ALTER TABLE on 10 billion rows: can take hours or days
  → Schema migrations become very risky operations

  None of these limits can be overcome with vertical scaling past a certain point —
  even the largest server available in the cloud has its own ceiling.

Sharding solves these problems by distributing data across several independent servers. Each shard has its own resources, so the load is distributed and no single point becomes the bottleneck.

Three Sharding Strategies #

Range-Based Sharding #

Range-based sharding divides data by ranges of shard key values. This is the easiest strategy to understand and implement.

Range-based sharding by user_id:

  Shard 1    │  Shard 2    │  Shard 3    │  Shard 4
  ───────────┼─────────────┼─────────────┼───────────
  id 1–25M   │  id 25M–50M │  id 50M–75M │  id 75M–100M

  Routing logic in the application:
  def get_shard(user_id):
      if user_id <= 25_000_000:   return shard_1
      elif user_id <= 50_000_000: return shard_2
      elif user_id <= 75_000_000: return shard_3
      else:                       return shard_4

// A simple shard router implementation for range-based sharding
type ShardRange struct {
    start int
    end   int
    name  string
}

type RangeShardRouter struct {
    shardRanges []ShardRange
    connections map[string]Connection
}

func (r *RangeShardRouter) GetShard(userID int) (string, error) {
    for _, sr := range r.shardRanges {
        if sr.start <= userID && userID < sr.end {
            return sr.name, nil
        }
    }
    return "", fmt.Errorf("no shard found for user_id: %d", userID)
}

func (r *RangeShardRouter) GetConnection(userID int) (Connection, error) {
    shardName, err := r.GetShard(userID)
    if err != nil {
        return Connection{}, err
    }
    conn, ok := r.connections[shardName]
    if !ok {
        return Connection{}, fmt.Errorf("no connection for shard: %s", shardName)
    }
    return conn, nil
}

// Queries are automatically routed to the right shard
func GetUserOrders(router *RangeShardRouter, userID int) (Result, error) {
    conn, err := router.GetConnection(userID)
    if err != nil {
        return Result{}, err
    }
    return conn.Query("SELECT * FROM orders WHERE user_id = ?", userID)
}

The main problem with range-based sharding: hot shards. New data almost always has large IDs — meaning the last shard always receives more writes than the earlier ones. This creates a load imbalance that gets worse over time.

The hot shard problem in range-based sharding:

  Time T1:              Time T2 (6 months later):
  Shard 1 [████████]     Shard 1 [████████]      ← unchanged, old data
  Shard 2 [██████]       Shard 2 [██████]         ← almost unchanged
  Shard 3 [████]         Shard 3 [████]
  Shard 4 [██]           Shard 4 [████████████████] ← HOT! all new users land here

  Shard 4 receives 90% of writes and 60% of reads → uneven load
  Solution: periodic resharding, or move to hash-based sharding

Hash-Based Sharding #

Hash-based sharding determines the shard from the hash value of the shard key. Distribution is far more even than range-based, because hash values are randomly distributed.

// Hash-based shard router
import (
    "crypto/md5"
    "encoding/binary"
    "fmt"
)

type HashShardRouter struct {
    numShards   int
    connections map[string]Connection
}

func (r *HashShardRouter) GetShardIndex(shardKey any) int {
    // Consistent and deterministic hashing
    keyBytes := []byte(fmt.Sprint(shardKey))
    sum := md5.Sum(keyBytes)
    return int(binary.BigEndian.Uint64(sum[:8]) % uint64(r.numShards))
}

func (r *HashShardRouter) GetConnection(shardKey any) Connection {
    index := r.GetShardIndex(shardKey)
    return r.connections[fmt.Sprintf("shard_%d", index)]
}

// Queries are automatically routed to the right shard
func GetUser(router *HashShardRouter, userID int) (Result, error) {
    conn := router.GetConnection(userID)
    return conn.Query("SELECT * FROM users WHERE id = ?", userID)
}

Hash-based sharding distribution (8 shards):

  user_id=1     → hash → shard_3  [████████]
  user_id=2     → hash → shard_7  [████████]
  user_id=3     → hash → shard_1  [████████]
  user_id=4     → hash → shard_5  [████████]
  ...all shards get roughly the same load

  No hot shards — even distribution regardless of insert patterns

The big weakness of hash-based sharding: resharding is very expensive. When you add a new shard, going from 8 to 16, almost all data must be moved because hash(key) % 8 and hash(key) % 16 give different results for almost every key.

The solution to this problem is consistent hashing — a technique that minimizes the amount of data that must be moved when shards are added or removed.

Consistent hashing — more efficient resharding:

  Hash ring (0–100):
  Shard A: range 0–24
  Shard B: range 25–49
  Shard C: range 50–74
  Shard D: range 75–100

  Adding shard E in the 50–62 range:
  Shard C: range 63–74 (shrunk, some data moves to E)
  Shard E: range 50–62 (new, takes some data from C)
  Shards A, B, D: unchanged

  Only ~1/N of the data must be moved (N = number of shards)
  compared to simple hashing which must move almost all data.

Directory-Based Sharding #

Directory-based sharding uses a lookup table (shard directory) storing the mapping between the shard key and the shard holding that data. This is the most flexible approach — you can move data between shards any time just by updating an entry in the directory.

// Directory-based shard router
type DirectoryShardRouter struct {
    directoryDB Connection
    localCache  map[int]string // local cache to avoid a lookup per request
}

func (r *DirectoryShardRouter) GetShard(tenantID int) (string, error) {
    // Check the local cache first
    if shard, ok := r.localCache[tenantID]; ok {
        return shard, nil
    }

    // Look up the shard directory
    result := r.directoryDB.Query(
        "SELECT shard_name FROM shard_directory WHERE tenant_id = ?", tenantID,
    )
    if len(result) == 0 {
        return "", fmt.Errorf("no shard found for tenant_id: %d", tenantID)
    }

    shardName := result[0]["shard_name"]
    r.localCache[tenantID] = shardName // store in cache
    return shardName, nil
}

// Assign a tenant to a specific shard — usually when onboarding a new tenant
func (r *DirectoryShardRouter) AssignShard(tenantID int, shardName string) {
    r.directoryDB.Execute(
        "INSERT INTO shard_directory (tenant_id, shard_name) VALUES (?, ?)", tenantID, shardName,
    )
    r.localCache[tenantID] = shardName
}

// Move a tenant from the old shard to a new shard without downtime
func (r *DirectoryShardRouter) MigrateTenant(tenantID int, newShard string) {
    // 1. Copy data to the new shard (background process)
    // 2. Update the directory (atomic)
    // 3. Invalidate the cache
    r.directoryDB.Execute(
        "UPDATE shard_directory SET shard_name = ? WHERE tenant_id = ?", newShard, tenantID,
    )
    delete(r.localCache, tenantID)
}

The shard_directory table:

  tenant_id │ shard_name  │ migrated_at
  ──────────┼─────────────┼─────────────
  101       │ shard_1     │ 2024-01-15
  102       │ shard_1     │ 2024-01-16
  103       │ shard_2     │ 2024-02-01  ← a large tenant moved to its own shard
  104       │ shard_1     │ 2024-02-03
  105       │ shard_3     │ 2024-03-10  ← an enterprise tenant gets a dedicated shard

Directory-based sharding offers the highest flexibility: you can move a single tenant from a full shard to a new shard just by updating one row in the directory, without changing the shard key or overhauling the architecture. This is very useful for multi-tenant SaaS where per-tenant load varies widely.

Its weakness: the shard directory is a new single point of failure. If the directory goes down, the whole system can’t determine which shard to access. Make sure the directory uses replication and aggressive caching.

Real Problems Sharding Brings #

Sharding solves scalability problems, but it creates new problems that don’t exist in single-database systems. Understanding these problems before implementing sharding is critical.

Cross-Shard Queries: The Most Expensive Operation #

Queries needing data from more than one shard are very expensive operations in a sharded system. There’s no such thing as a cross-shard JOIN at the database level — everything must be done in the application layer, meaning multiple round-trips to the database and in-memory aggregation.

// Cross-shard queries — very expensive and hard to scale
func getAllUsersWithOrders(dateRange DateRange) ([]QueryResult, error) {
    var results []QueryResult

    // Must query every shard one by one (or in parallel)
    for _, shard := range []Shard{shard1, shard2, shard3, shard4} {
        partialResults, err := shard.Query(
            `SELECT u.id, u.name, COUNT(o.id) as order_count
FROM users u
LEFT JOIN orders o ON o.user_id = u.id
WHERE o.created_at BETWEEN ? AND ?
GROUP BY u.id`,
            dateRange.Start, dateRange.End,
        )
        if err != nil {
            return nil, err
        }
        results = append(results, partialResults...)
    }

    // Combine and sort in the application layer
    sort.Slice(results, func(i, j int) bool {
        return results[i]["order_count"] > results[j]["order_count"]
    })
    return results, nil
}

// Problems:
// → Latency = max(latency of all shards) — can't be faster than the slowest shard
// → Memory: all results from all shards must fit in application memory
// → Can't LIMIT at the database level — must fetch everything then filter in the app
// → Hard to paginate correctly

The most important implication of this problem: the shard key must be chosen based on the most frequent query patterns. If 95% of queries can be completed within one shard, the system remains very efficient. If 30% of queries need to cross shards, the system will be very slow and hard to scale.

Distributed Transactions: Lost Consistency #

In a single database, transactions guarantee atomicity — all or nothing. In a sharded system, transactions involving more than one shard need distributed transactions, which are far more complex.

A problematic distributed transaction scenario:

  Transferring a balance between user A (shard 1) and user B (shard 3):

  Step 1: Decrease user A's balance in shard_1
  Step 2: Increase user B's balance in shard_3

  What can go wrong:
  → Step 1 succeeds, but Step 2 fails:
    user A loses money, user B receives nothing → money disappears
  → Step 1 succeeds, the connection to shard_3 breaks:
    same — money disappears without a clear trace
  → Both shards are slow → timeout at the coordinator → unclear which one succeeded

Solutions for distributed transactions:

Option 1: Two-Phase Commit (2PC)
  → The coordinator asks all shards: "ready to commit?"
  → If all are ready, the coordinator orders the commit
  → If any isn't ready, the coordinator orders a rollback for all
  → Problem: the coordinator can fail between prepare and commit
    (leaving shards in an uncertain state)
  → Very slow and not scalable for high traffic

Option 2: The Saga Pattern
  → Each step is done separately
  → If a step fails, run compensating transactions
  → Example: if Step 2 fails, restore user A's balance
  → More complex but more resilient than 2PC
  → Eventual consistency — there's a time window where state is inconsistent

Option 3: Redesign so transactions stay within one shard
  → This is the best solution
  → For example: store the balance in the shard determined by the sender's user_id
  → A transfer is recorded as a "debit from A" in shard A and a "credit to B" in shard B
  → Two separate records, no atomic cross-shard transaction needed

Never implement distributed transactions without fully understanding their consequences for data consistency. Partial failure in a distributed transaction is more dangerous than total failure — because some data changes and some doesn’t, and there’s no automatic rollback. Designing a shard key that minimizes cross-shard transaction needs is far better than relying on distributed transactions.

Resharding: The Most Feared Operation #

One day, the number of shards you have is no longer enough. One or several shards start filling up. You need to add new shards and redistribute the data into them. This is called resharding — and it’s one of the most complex and risky operations in a sharded system.

Resharding challenges:

  Before: 4 shards → After: 8 shards

  Simple hash (hash % 4 → hash % 8):
  → Almost all data lands in a different shard position
  → Must copy ~75% of the total data
  → Long downtime or dual-write period
  → Consistency risk during the migration process

  Consistent hashing:
  → Only ~12.5% of the data must be moved (1/8 of the total)
  → Far faster and safer
  → But the implementation is more complex

  A relatively safe resharding strategy:
  Step 1: Create the new shards but keep them inactive
  Step 2: Start dual-writes — write to the old shard AND the new shard
  Step 3: Backfill — copy old data to the new shards in the background
  Step 4: Verify consistency between the old and new shards
  Step 5: Switch traffic to the new shards
  Step 6: Delete data from the old shards after a stabilization period

Poorly planned resharding can cause hours of downtime or even data loss. This is why the resharding strategy must be designed from the first sharding implementation — not when you’re already under pressure.

Choosing the Right Shard Key #

Shard key selection is the most important decision in sharding implementation — and the hardest to change once the system is running. A wrong shard key can erase all of sharding’s benefits and add complexity without value.

Criteria for a good shard key:

  ✓ High cardinality
    → Many distinct values ensure even distribution
    → user_id, order_id, transaction_id → good
    → status, country_code → too few values, poor distribution

  ✓ Even distribution
    → Values distributed evenly across all shards
    → user_id with auto-increment + hash → even
    → created_at without hashing → uneven (hot shard at the newest values)

  ✓ Always present in the most frequent queries
    → If 90% of queries filter by user_id, use user_id
    → If the shard key isn't in the WHERE clause → cross-shard query

  ✓ Immutable — never changes
    → If the shard key changes, data must move between shards
    → user_id doesn't change → safe
    → email can change → dangerous as a shard key

  ✓ Enough granularity for future distribution
    → If there are 10 shards and the shard key only has 10 unique values,
      the 11th shard can't be added

Good shard key columns:
  user_id, customer_id, tenant_id, account_id

Bad shard key columns:
  created_at (hot shards), status (low cardinality),
  email (can change), country_code (only a few values)

The Complete Sharding Architecture #

In production, sharding rarely stands alone. It’s combined with replication within each shard for high availability, partitioning for data management within shards, and connection poolers for connection efficiency.

The complete sharding architecture in production:

flowchart TD
    Router["Shard Router (determines the shard based on the shard key)"]

    subgraph Shard1["Shard 1"]
        direction TB
        P1["Primary"]
        R1_1["Replica 1"]
        R1_2["Replica 2"]
        Part1["Monthly partitions"]

        P1 -. replication .-> R1_1
        P1 -. replication .-> R1_2
    end

    subgraph Shard2["Shard 2"]
        direction TB
        P2["Primary"]
        R2_1["Replica 1"]
        R2_2["Replica 2"]
        Part2["Monthly partitions"]

        P2 -. replication .-> R2_1
        P2 -. replication .-> R2_2
    end

    subgraph Shard3["Shard 3"]
        direction TB
        P3["Primary"]
        R3_1["Replica 1"]
        R3_2["Replica 2"]
        Part3["Monthly partitions"]

        P3 -. replication .-> R3_1
        P3 -. replication .-> R3_2
    end

    Router --> Shard1
    Router --> Shard2
    Router --> Shard3

Each shard: → Has its own primary + replicas (high availability) → Large tables partitioned by time (query efficiency) → A connection pooler in front (connection efficiency)

Anti-Patterns to Avoid #

✗ Anti-pattern 1: sharding before its time
  A system with 50 million rows and 1,000 active users doesn't need sharding.
  Query optimization, indexes, partitioning, and read replicas are far more appropriate.
  Premature sharding adds complexity without real benefit.
  ✓ Exhaust all simpler techniques before considering sharding.

────────────────────────────────────────────────────────────────────────────────

✗ Anti-pattern 2: a shard key absent from queries
  If 50% of queries don't include the shard key in the WHERE clause,
  those 50% will always be cross-shard queries.
  A heavily sharded system can be slower than a single database.
  ✓ Analyze query patterns deeply before choosing a shard key.
    Make sure the shard key appears in the majority of the most frequent queries.

────────────────────────────────────────────────────────────────────────────────

✗ Anti-pattern 3: relying on cross-shard JOINs
  Cross-shard JOINs don't exist at the database level — they must be done in the app.
  This is slow, consumes lots of memory, and can't be paginated correctly.
  ✓ Design the data model so important queries can be completed within one shard.
    Denormalize if needed to avoid cross-shard joins.

────────────────────────────────────────────────────────────────────────────────

✗ Anti-pattern 4: not preparing a resharding strategy
  "We start with 4 shards; later when they fill up, we'll add more."
  Without a clear resharding plan, adding shards becomes a very risky
  operation that can cause long downtime.
  ✓ Design the resharding strategy from the start. Consider consistent hashing
    or directory-based sharding for resharding flexibility.

────────────────────────────────────────────────────────────────────────────────

✗ Anti-pattern 5: hardcoding the shard count in application code
  if user_id % 4 == 0: use shard_1  # hardcoded 4 shards!
  When adding the 5th shard, every piece of code containing the number 4 must change.
  ✓ Abstract shard routing into one centralized component.
    The shard count must be changeable without changing business logic.

────────────────────────────────────────────────────────────────────────────────

✗ Anti-pattern 6: ignoring hot shards
  Range-based sharding on sequential IDs → the last shard is always hottest.
  No monitoring detects this → one shard overloaded while others sit idle —
  completely defeating the purpose of sharding.
  ✓ Monitor per-shard load distribution. If uneven, consider
    moving to hash-based sharding or adding new shards with resharding.

Sharding Review Checklist #

SHARDING DECISIONS:
  □ All simpler techniques already optimized and found insufficient
    (query optimization, indexes, partitioning, read replicas, caching)
  □ Data volume or write throughput already beyond a single server's capacity
  □ The team has the capability to manage a far more complex system
  □ A plan exists for handling cross-shard queries and distributed transactions

SHARD KEY SELECTION:
  □ Shard key chosen based on query pattern analysis, not intuition
  □ Shard key present in the majority of the most frequent queries
  □ Shard key has high cardinality and even distribution
  □ Shard key immutable — its value never changes after insert
  □ Shard key provides enough granularity for future shard additions

IMPLEMENTATION:
  □ Shard routing abstracted into one centralized component
  □ Shard count not hardcoded in business logic
  □ Cross-shard queries identified with a handling plan
  □ Distributed transactions minimized or replaced with the Saga pattern
  □ Every shard has replication for high availability

RESHARDING:
  □ A resharding strategy designed from the start
  □ The resharding process documented and tested in staging
  □ Consistent hashing or directory-based sharding considered
    to ease future resharding

OBSERVABILITY:
  □ Per-shard metrics available: QPS, latency, disk usage, connection counts
  □ Per-shard data distribution monitored — alert if uneven (hot shards)
  □ Cross-shard queries identified and monitored
  □ Per-shard error rates monitored — alert if one shard has more errors

Summary #

• Sharding is the last step, not the first — it solves problems no other approach can, but at a very high complexity cost. Exhaust all simpler techniques before considering sharding.
• Sharding differs from partitioning — partitioning works within one database instance (one server), sharding works at the infrastructure level (several independent servers). The two are often combined.
• The shard key is the most important decision — and the hardest to change. It must have high cardinality, even distribution, always appear in the most frequent queries, and be immutable.
• Range-based sharding is easy to understand but prone to hot shards — the shard holding the newest data is always hotter. Consider hash-based or consistent hashing for more even distribution.
• Hash-based sharding distributes evenly but makes resharding expensive — adding a new shard forces most data to move. Consistent hashing minimizes the data that must move.
• Directory-based sharding is the most flexible — tenants can be moved between shards at any time, but the shard directory is a single point of failure whose availability must be protected.
• Cross-shard queries must be avoided as much as possible — they can’t be done at the database level; they must be done in the application layer with multiple round-trips, in-memory aggregation, and pagination difficulties.
• Distributed transactions are far more dangerous than they look — partial failures can leave data in an inconsistent state without automatic rollback. Design so transactions stay within one shard whenever possible.
• Resharding must be designed from the start — the most feared operation in a sharded system. Without a clear plan, adding new shards can become very long downtime.
• Per-shard observability is mandatory — without load distribution monitoring, a hot shard killing one node won’t be detected until an incident happens.