Database Sharding

Scale databases horizontally by distributing data across multiple shards. Covers sharding strategies, shard key selection, cross-shard queries, rebalancing, and the patterns that manage the complexity of distributed data.

Database sharding is the process of splitting a large database into smaller, manageable pieces called shards, each stored on its own database server or cluster. It matters when a single database instance cannot handle the volume of data, queries, or concurrent connections—when read/write latency degrades, replication lag increases, or a single node becomes a bottleneck. Sharding enables horizontal scaling, distributing data and load across multiple machines without requiring application changes to access the full dataset.

Choosing a Shard Key

The shard key determines how data is partitioned across shards. It must be chosen carefully: a poor choice leads to uneven load, hot shards, and inefficient queries.

The shard key is a column or set of columns used to compute a shard ID, typically by applying a hashing function or range-based partitioning. A common choice is a primary key or a frequently queried field such as user_id, tenant_id, or created_at.

For example, in a multi-tenant SaaS application, tenant_id is a natural shard key. Each tenant’s data lives on one or more shards, and all queries for a tenant can be routed to that shard.

-- Example: shard by tenant_id using modulo hashing
SELECT
  tenant_id,
  id,
  name,
  created_at
FROM users
WHERE tenant_id = 123;

The shard key must be part of the primary key or indexed on every shard. Queries that do not include the shard key in their WHERE clause may require a scatter-gather operation across all shards.

A common mistake is using a non-uniform distribution of shard key values. If tenant_id values are skewed—e.g., 80% of users belong to ten tenants—then those ten tenants become hot shards, while others remain underutilized.

To detect this, query the shard distribution:

-- On each shard, run:
SELECT
  tenant_id,
  COUNT(*) AS user_count
FROM users
GROUP BY tenant_id
ORDER BY user_count DESC
LIMIT 10;

If the top 10 tenants account for more than 50% of total rows, consider re-sharding using a composite key or a consistent hashing scheme.

Mapping Data to Shards

Once the shard key is selected, data must be distributed using a consistent mapping function. The most common strategies are:

Hash-based sharding is simple and balances data well but requires a consistent hashing algorithm for dynamic shard addition.

-- Example: compute shard ID using MurmurHash3
WITH shard_mapping AS (
  SELECT
    tenant_id,
    (MurmurHash3(tenant_id) % 16) AS shard_id
  FROM tenants
)
SELECT
  t.tenant_id,
  t.name,
  s.shard_id,
  CONCAT('db-', s.shard_id, '.example.com') AS shard_host
FROM tenants t
JOIN shard_mapping s ON t.tenant_id = s.tenant_id;

In practice, this mapping is often stored in a lookup table:

-- Shard mapping table
CREATE TABLE shard_map (
  shard_key_value VARCHAR(255) PRIMARY KEY,
  shard_id INTEGER NOT NULL,
  shard_host VARCHAR(255) NOT NULL,
  last_updated TIMESTAMPTZ DEFAULT NOW()
);

Application code reads from this table to route queries. A failure mode occurs when the mapping table becomes unavailable: the system falls back to a cached version, but stale mappings lead to routing errors.

A common failure is shard drift: data is written to a shard, but the mapping table is not updated. A new tenant is created and assigned to shard 5, but the shard_map entry is missing. Subsequent queries fail to find the tenant, resulting in a 500 error or a 404-like response.

Routing Queries to Shards

Application code or a middleware layer routes each query to the correct shard. The most common approach is to use a shard router—a service that reads the shard key from the query and forwards it to the appropriate database instance.

For a SELECT * FROM users WHERE tenant_id = 123, the router:

  1. Extracts tenant_id = 123 from the query.
  2. Looks up the shard ID for 123 in the shard_map table.
  3. Connects to db-5.example.com and executes the query.
  4. Returns results to the client.

This routing can be implemented in an application layer using a library such as pgx in Go:

func GetUserByTenant(db *pgx.Conn, tenantID int) (User, error) {
    var user User
    var shardID int

    // Query shard map
    err := db.QueryRow(`
        SELECT shard_id FROM shard_map WHERE shard_key_value = $1
    `, tenantID).Scan(&shardID)
    if err != nil {
        return user, fmt.Errorf("failed to get shard for tenant %d: %w", tenantID, err)
    }

    // Connect to shard
    shardConn, err := pgx.Connect(context.Background(), fmt.Sprintf(
        "host=db-%d.example.com user=app password=secret dbname=users",
        shardID,
    ))
    if err != nil {
        return user, fmt.Errorf("failed to connect to shard %d: %w", shardID, err)
    }
    defer shardConn.Close(context.Background())

    // Execute query on shard
    err = shardConn.QueryRow(`
        SELECT id, name, created_at FROM users WHERE tenant_id = $1
    `, tenantID).Scan(&user.ID, &user.Name, &user.CreatedAt)
    if err != nil {
        return user, fmt.Errorf("query failed on shard %d: %w", shardID, err)
    }

    return user, nil
}

When a query involves multiple tables or joins across shards, the router must gather results from multiple shards.

For a JOIN of users and orders, where both tables are sharded by tenant_id, the router issues separate queries to each shard and combines the results:

-- On each shard, run:
SELECT
  u.id AS user_id,
  u.name,
  o.id AS order_id,
  o.total
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE u.tenant_id = 123;

This results in a scatter-gather operation. A failure in any shard causes the entire query to fail. The most common error is a timeout: if one shard takes longer than others, the query waits for the slowest, causing overall latency spikes.

Handling Cross-Shard Queries

Not all queries can be confined to a single shard. Queries involving multiple tenants or global aggregations must span all shards.

A common cross-shard query is a global aggregation:

-- Total revenue across all tenants
SELECT
  SUM(total) AS total_revenue,
  AVG(total) AS avg_order_value
FROM orders;

This requires issuing the same query on every shard and combining results.

The router collects results as a stream of partial aggregates:

Shard 0: { total_revenue: 125000, avg_order_value: 142.5 }
Shard 1: { total_revenue: 231000, avg_order_value: 138.0 }
Shard 2: { total_revenue: 98000, avg_order_value: 155.0 }
...

It then computes the final result:

total_revenue = 125000 + 231000 + 98000 + ... = 784000
avg_order_value = (125000 + 231000 + 98000 + ...) / (125000/142.5 + 231000/138.0 + ...)

A failure mode occurs when a shard returns a NULL value for total_revenue. The router must handle NULLs correctly—e.g., by using COALESCE or by aggregating with SUM and COUNT separately.

For complex cross-shard queries, the application may pre-aggregate data using a materialized view or a data warehouse.

Adding and Removing Shards

Scaling the database requires adding new shards and rebalancing data. The process involves:

  1. Adding a new shard: create a new database instance, configure it, and update the shard_map table.
  2. Rebalancing: migrate data from existing shards to new ones.
  3. Rolling out: update application code to route new data to the new shard and gradually shift existing data.

Adding a shard is simple:


docker run -d --name db-16 \
  -e POSTGRES_DB=users \
  -e POSTGRES_USER=app \
  -e POSTGRES_PASSWORD=secret \
  -p 5433:5432 \
  postgres:15

# Register the shard in the mapping table
psql -h db-16.example.com -U app users -c "
  INSERT INTO shard_map (shard_key_value, shard_id, shard_host)
  VALUES ('16', 16, 'db-16.example.com');
"

Rebalancing data involves a migration process. A common approach is splitting an existing shard:

  1. Identify a large shard (e.g., shard_id = 5).
  2. Create a new shard (shard_id = 17).
  3. Copy data from shard_id = 5 to shard_id = 17, using a consistent snapshot.
  4. Update shard_map to route new data to both 5 and 17.
  5. Gradually move older data to 17.

During rebalancing, the system must handle dual writes: writes to both the old and new shard.

-- During rebalancing: write to both shard 5 and shard 17
INSERT INTO users (tenant_id, name, created_at)
VALUES (123, 'Alice', NOW());

-- Also write to the new shard
INSERT INTO users (tenant_id, name, created_at)
VALUES (123, 'Alice', NOW())
ON SHARD 17;

A failure here is write skew: a tenant is updated on shard 5, but the update is not yet reflected on shard 17. The application reads stale data.

Rebalancing with Dual Writes

To reduce skew, use a write-through strategy:

The key challenge is ensuring consistency during the transition. A common failure is missing writes: a new tenant is created and written to shard 5, but the shard_map entry is missing, so future reads miss the tenant entirely.

Failure Modes and Monitoring

Sharding introduces new failure modes not present in monolithic databases:

Monitor these with:

A critical alert is when a shard fails to respond to a health check for more than 30 seconds. The system should automatically fail over to a standby or mark the shard as offline.

A common error in logs is:

ERROR: shard 'db-5.example.com' not found in shard_map
  context: tenant_id = 123, shard_id = 5
  stack trace: ...

This indicates that a tenant was created, but the shard map entry was missing.

Summary

Database sharding is the process of distributing data across multiple database instances based on a shard key. It enables horizontal scaling and improves performance for large, high-traffic systems. The core components are choosing a shard key, mapping data to shards, routing queries, handling cross-shard operations, and managing shard lifecycle. The most common failure modes stem from inconsistent mappings, unavailability, and skewed data distribution. Monitoring and operational practices must be tuned to detect and resolve these issues before they impact users.

This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.