Database Sharding: Horizontal Partitioning, Consistent Hashing, and Resharding

Database sharding is horizontal partitioning — splitting data across multiple database instances to distribute load that a single database can no longer handle. It's one of the most complex scaling decisions you can make, with significant operational cost and irreversible consequences if you pick the wrong shard key.

The first question is always: do you actually need sharding?

---

Before Sharding: Exhaust These First

Most systems never need sharding. PostgreSQL handles 100M–1B rows per table efficiently with proper indexing. Read replicas handle read-heavy workloads. Vertical scaling is cheap at modern cloud prices.

Shard only when you've exhausted these and you have a specific, measured bottleneck.

---

Sharding Concepts

Shard: A subset of data stored on a specific database node. Shard key: The field used to determine which shard a row belongs to. Shard function: Maps shard key values to shard nodes.

Data: 10M users
Shard key: user_id
Shard function: hash(user_id) % 4

user_id=1001 → hash=0x4f → shard 3
user_id=1002 → hash=0x2a → shard 2
user_id=1003 → hash=0x11 → shard 1

All data for a given user lives on a single shard — queries that filter by user_id hit one shard.

---

Shard Key Selection

The shard key is the most critical decision. A bad shard key causes:

• Hot spots: One shard gets 80% of traffic while others sit idle
• Cross-shard queries: Joins across shards require application-level scatter-gather
• Uneven data distribution: One shard grows faster than others

Good shard keys:

• High cardinality: Thousands+ of distinct values (not just 50 US states)
• Even distribution: Values spread relatively evenly across the range
• Query alignment: The most common queries filter by this key
• Immutable: Never changes after row creation (changing shard key requires moving data)

Common patterns:

---

Consistent Hashing

Naive modulo hashing (hash(key) % N) breaks when you add or remove shards — you'd need to move most of your data to different nodes.

Consistent hashing places shards on a "ring" so that adding/removing a shard only relocates ~1/N of the data, not all of it.

# consistent_hashing.py
import hashlib
from bisect import bisect_right, insort

class ConsistentHashRing:
    def __init__(self, virtual_nodes: int = 150):
        """
        virtual_nodes: Number of virtual nodes per physical node.
        More virtual nodes = more even distribution.
        """
        self.virtual_nodes = virtual_nodes
        self.ring: list[int] = []  # Sorted list of positions
        self.node_map: dict[int, str] = {}  # position → node name

    def add_node(self, node: str):
        for i in range(self.virtual_nodes):
            key = f"{node}:vnode:{i}"
            position = int(hashlib.md5(key.encode()).hexdigest(), 16)
            insort(self.ring, position)
            self.node_map[position] = node

    def remove_node(self, node: str):
        for i in range(self.virtual_nodes):
            key = f"{node}:vnode:{i}"
            position = int(hashlib.md5(key.encode()).hexdigest(), 16)
            self.ring.remove(position)
            del self.node_map[position]

    def get_node(self, key: str) -> str:
        """Find which node is responsible for a given key"""
        if not self.ring:
            raise ValueError("No nodes in ring")

        key_position = int(hashlib.md5(key.encode()).hexdigest(), 16)

        # Find first node clockwise from key_position
        idx = bisect_right(self.ring, key_position) % len(self.ring)
        return self.node_map[self.ring[idx]]

# Usage
ring = ConsistentHashRing(virtual_nodes=150)
ring.add_node("shard-1")
ring.add_node("shard-2")
ring.add_node("shard-3")
ring.add_node("shard-4")

# Route a query
user_id = "user-abc-123"
target_shard = ring.get_node(user_id)
print(f"User {user_id} → {target_shard}")

# Add a 5th shard — only ~20% of keys move, not 80%
ring.add_node("shard-5")

Virtual nodes prevent hot spots when physical nodes have different sizes or when you have few shards. With 150 virtual nodes per physical node, data distributes evenly even with 4–5 shards.

---

PostgreSQL Table Partitioning (Before Full Sharding)

PostgreSQL's native partitioning splits a large table across partitions within a single database — much simpler than true sharding and often sufficient.

-- Range partitioning by date (common for time-series data)
CREATE TABLE events (
    id          BIGSERIAL,
    user_id     UUID NOT NULL,
    event_type  TEXT NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    payload     JSONB
) PARTITION BY RANGE (created_at);

-- Create monthly partitions
CREATE TABLE events_2026_01 PARTITION OF events
    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');

CREATE TABLE events_2026_02 PARTITION OF events
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');

-- Index on each partition (automatically propagates)
CREATE INDEX ON events (user_id, created_at);

-- PostgreSQL routes queries to correct partition automatically
EXPLAIN SELECT * FROM events
WHERE created_at BETWEEN '2026-01-01' AND '2026-01-31'
  AND user_id = 'abc-123';
-- → Scans only events_2026_01, not all other partitions

-- Automate partition creation (run monthly via cron)
CREATE OR REPLACE FUNCTION create_next_month_partition()
RETURNS void AS $$
DECLARE
    next_month DATE := DATE_TRUNC('month', NOW() + INTERVAL '1 month');
    partition_name TEXT := 'events_' || TO_CHAR(next_month, 'YYYY_MM');
BEGIN
    EXECUTE format(
        'CREATE TABLE IF NOT EXISTS %I PARTITION OF events
         FOR VALUES FROM (%L) TO (%L)',
        partition_name,
        next_month,
        next_month + INTERVAL '1 month'
    );
END;
$$ LANGUAGE plpgsql;

---

Application-Level Sharding Router

// lib/shardRouter.ts
import { Pool } from 'pg';
import { ConsistentHashRing } from './consistentHash';

const SHARD_CONNECTIONS: Record<string, Pool> = {
  'shard-1': new Pool({ connectionString: process.env.SHARD_1_URL }),
  'shard-2': new Pool({ connectionString: process.env.SHARD_2_URL }),
  'shard-3': new Pool({ connectionString: process.env.SHARD_3_URL }),
  'shard-4': new Pool({ connectionString: process.env.SHARD_4_URL }),
};

const ring = new ConsistentHashRing(150);
Object.keys(SHARD_CONNECTIONS).forEach(shard => ring.addNode(shard));

export function getShardForUser(userId: string): Pool {
  const shardName = ring.getNode(userId);
  return SHARD_CONNECTIONS[shardName];
}

// Queries automatically route to the right shard
export async function getUserById(userId: string) {
  const db = getShardForUser(userId);
  const result = await db.query(
    'SELECT * FROM users WHERE id = $1',
    [userId]
  );
  return result.rows[0];
}

// Cross-shard query: requires scatter-gather
export async function getUsersByEmail(email: string) {
  // Must query ALL shards and merge results
  const promises = Object.values(SHARD_CONNECTIONS).map(db =>
    db.query('SELECT * FROM users WHERE email = $1', [email])
  );
  const results = await Promise.all(promises);
  return results.flatMap(r => r.rows);
}

---

Resharding: Adding Shards Without Downtime

When a shard grows too large, you need to split it. With consistent hashing, this involves:

1. Create the new shard node
2. Add it to the ring (some key ranges migrate to it)
3. Migrate data: read from old shard, write to new shard
4. Cut over traffic once migration is complete
5. Delete migrated data from old shard

# resharding.py — migrate data between shards
async def migrate_shard_data(
    source_shard: Pool,
    destination_shard: Pool,
    ring: ConsistentHashRing,
    new_node: str,
    batch_size: int = 1000,
):
    """Migrate rows that now belong to new_node"""
    offset = 0

    while True:
        # Fetch batch from source
        rows = await source_shard.fetch(
            f"SELECT * FROM users LIMIT {batch_size} OFFSET {offset}"
        )
        if not rows:
            break

        # Filter rows that now belong to the new node
        migrated = [
            row for row in rows
            if ring.get_node(str(row['user_id'])) == new_node
        ]

        if migrated:
            # Bulk insert to destination
            await destination_shard.executemany(
                "INSERT INTO users VALUES ($1, $2, ...) ON CONFLICT DO NOTHING",
                [(row['user_id'], row['email'], ...) for row in migrated]
            )

        offset += batch_size
        # Rate limit to avoid overloading production
        await asyncio.sleep(0.01)

Resharding is operationally expensive. Double-write patterns (write to both old and new shard during migration) ensure no data loss. Plan for resharding to take days, not hours, on large datasets.

---

Working With Viprasol

We design database architectures that grow with your product — from proper indexing and partitioning through read replicas and, when truly necessary, sharding. Our work includes both the data model design and the infrastructure setup.

→ Talk to our database team about scaling your data layer.

---

See Also

• PostgreSQL Performance Optimization — indexes, EXPLAIN ANALYZE, before sharding
• Redis Use Cases — caching layer to reduce shard load
• Data Engineering Pipeline — ETL/ELT for analytical workloads
• Database Connection Pooling — PgBouncer before scaling out
• Cloud Solutions — database architecture and infrastructure