The Runtime Theory
Databases

Horizontal Partitioning and Shard Keys

Range vs hash partitioning, shard key selection, and the hidden complexities of distributing data across nodes.

The Runtime Theory Team12 min read#partitioning#sharding#scalability#data-distribution
▸ On this page

When a single database can't handle your data volume or write throughput, horizontal partitioning splits your data across multiple nodes. But the choice of partitioning strategy and shard key determines whether your system scales linearly or creates hotspots that bottleneck the entire cluster.

Partition Types

PostgreSQL supports native table partitioning. The data is split into child tables that inherit from a parent table, and the planner routes queries to the correct partition.

Range Partitioning

Range partitioning divides data by contiguous ranges of a key:

sql
-- Range partitioning by time
CREATE TABLE events (
    id bigserial,
    created_at timestamptz NOT NULL,
    user_id bigint,
    payload jsonb
) PARTITION BY RANGE (created_at);
 
CREATE TABLE events_2026_q1 PARTITION OF events
    FOR VALUES FROM ('2026-01-01') TO ('2026-04-01');
 
CREATE TABLE events_2026_q2 PARTITION OF events
    FOR VALUES FROM ('2026-04-01') TO ('2026-07-01');
 
CREATE TABLE events_2026_q3 PARTITION OF events
    FOR VALUES FROM ('2026-07-01') TO ('2026-10-01');
 
CREATE TABLE events_2026_q4 PARTITION OF events
    FOR VALUES FROM ('2026-10-01') TO ('2027-01-01');
 
-- Query automatically routes to correct partition
EXPLAIN SELECT * FROM events WHERE created_at = '2026-05-15';
-- Query on events_2026_q2 only

Range partitioning excels for time-series data where queries typically filter by time range. Old partitions can be detached and archived cheaply.

Hash Partitioning

Hash partitioning distributes rows evenly across partitions using a hash function:

sql
-- Hash partitioning by user_id
CREATE TABLE user_data (
    user_id bigint,
    data jsonb,
    updated_at timestamptz
) PARTITION BY HASH (user_id);
 
CREATE TABLE user_data_p0 PARTITION OF user_data
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE user_data_p1 PARTITION OF user_data
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE user_data_p2 PARTITION OF user_data
    FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE user_data_p3 PARTITION OF user_data
    FOR VALUES WITH (MODULUS 4, REMAINDER 3);

Hash partitioning ensures even distribution but loses the ability to do range queries efficiently across partitions.

Shard Keys: The Most Important Decision

A shard key determines how data is distributed across database nodes. The choice affects:

  1. Distribution: Is data spread evenly or concentrated on one node?
  2. Query routing: Can you send queries to the right node without scattering?
  3. Hotspots: Does one shard handle disproportionate traffic?
python
# Shard key examples
 
# Good: user_id for a multi-tenant SaaS
# Queries always include user_id → single shard access
def get_user_data(user_id, shard_key=None):
    shard = hash(user_id) % NUM_SHARDS
    return db.execute(
        f"SELECT * FROM data WHERE user_id = %s",
        (user_id,)
    )
 
# Bad: timestamp for an event system
# Recent events concentrate on one shard (hotspot)
# Historical queries scatter across all shards
def get_events(timestamp_range):
    # Must query all shards for any time range
    results = []
    for shard in range(NUM_SHARDS):
        results.extend(query_shard(shard, timestamp_range))
    return results

Consistent Hashing for Dynamic Shards

When adding or removing nodes, consistent hashing minimizes data movement:

python
import hashlib
from bisect import bisect_right
 
class ConsistentHashRing:
    def __init__(self, nodes, virtual_nodes=150):
        self.ring = {}
        self.sorted_keys = []
        
        for node in nodes:
            for i in range(virtual_nodes):
                key = self._hash(f"{node}:{i}")
                self.ring[key] = node
                self.sorted_keys.append(key)
        
        self.sorted_keys.sort()
    
    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def get_node(self, data_key):
        hash_val = self._hash(data_key)
        idx = bisect_right(self.sorted_keys, hash_val) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]
    
    def add_node(self, node):
        for i in range(150):
            key = self._hash(f"{node}:{i}")
            self.ring[key] = node
            self.sorted_keys.append(key)
        self.sorted_keys.sort()
        # Only ~1/N of keys need to move
    
    def remove_node(self, node):
        for i in range(150):
            key = self._hash(f"{node}:{i}")
            del self.ring[key]
            self.sorted_keys.remove(key)

tradeoff / Distribution Evenness vs Query Efficiency

Choose your shard key based on your most common query pattern. If 90% of queries include user_id, shard by user_id. The remaining cross-shard queries are the price you pay.

Hash keys distribute evenly but scatter range queries. Range keys enable efficient range scans but risk hotspots. Compound keys (user_id, timestamp) combine both benefits for specific access patterns.

Cross-Shard Queries

The hardest problem in sharded databases is queries that span multiple shards:

sql
-- Cross-shard aggregation: requires scatter-gather
-- Application must query all shards and merge results
 
-- Pseudo-code for cross-shard aggregation
async function getTopUsers(limit):
    results = []
    for shard in shards:
        shard_results = await shard.query(
            "SELECT user_id, COUNT(*) as events FROM events GROUP BY user_id LIMIT $1",
            [limit]
        )
        results.extend(shard_results)
    
    # Merge and re-aggregate in application
    merged = aggregate_by_user_id(results)
    return merged.sort_by_events_desc().limit(limit)

Citus (PostgreSQL extension) handles this transparently:

sql
-- Citus distributed table
CREATE TABLE events (
    id bigserial,
    created_at timestamptz,
    user_id bigint,
    payload jsonb
);
 
SELECT create_distributed_table('events', 'user_id');
 
-- Citus routes queries automatically
SELECT user_id, COUNT(*)
FROM events
WHERE created_at > '2026-01-01'
GROUP BY user_id
ORDER BY COUNT(*) DESC
LIMIT 10;
-- Citus sends to all shards, merges results

Rebalancing: The Hidden Cost

Adding a node to a sharded cluster requires rebalancing — moving data from existing nodes to the new one. This is the hidden cost of sharding.

python
# Rebalancing strategies
 
# 1. Full rehash (Citus default)
# Rehash all keys, move affected rows
# Pro: Even distribution after rebalance
# Con: Moves ~1/N of all data (10% for 10 nodes)
 
# 2. Range splitting
# Split the largest range into two
# Pro: Only moves data from one node
# Con: May not achieve perfect balance
 
# 3. Consistent hashing
# Only keys between old and new node move
# Pro: Minimal data movement (~1/N)
# Con: Requires virtual nodes for even distribution
 
# During rebalancing, queries may be slower
# Some systems use a "shadow" mode:
# 1. Start writing to new node
# 2. Background-copy existing data
# 3. Switch reads to new node
# 4. Stop writing to old node

Synthesis

Horizontal partitioning and sharding are powerful tools for scaling beyond a single database. The shard key is the most critical decision — it determines distribution, query routing, and hotspot potential. Range partitioning suits time-series data; hash partitioning suits point-lookup workloads. Cross-shard queries remain the fundamental challenge, requiring either application-level merging or distributed query engines like Citus. Choose your strategy based on your access patterns, not your data model.

One dispatch a week

If this kind of tracing interests you, subscribe. One technical explanation per week — no noise, no course spam.

One technical dispatch per week. No noise.