Distributed Database Patterns

Distributed databases span multiple nodes across networks, enabling high availability, scalability, and fault tolerance. This page describes the core patterns used to design, operate, and troubleshoot such systems, focusing on the practical mechanics of achieving consistency, partition tolerance, and performance across geographies and failure domains.

Data Placement and Distribution

When data grows beyond a single node’s capacity or latency budget, the system must distribute it across machines. The choice of distribution strategy directly affects query performance, update consistency, and failure recovery.

Sharding by Key Range

Data is partitioned into contiguous ranges of a key, typically the primary key. Each shard holds a subset of rows where key ≥ start and key < end. A router service or client library maps incoming queries to shards using a lookup table or metadata cache.

Shards are assigned to database nodes, and a coordinator — often a stateless frontend — routes queries. For example, a request to GET /users/1000 is routed to the node hosting the shard where 1000 falls.

-- Shard assignment metadata stored in a system table
CREATE TABLE shards (
    shard_id INT PRIMARY KEY,
    start_key BIGINT,
    end_key BIGINT,
    node_address TEXT,
    status TEXT CHECK (status IN ('active', 'draining', 'readonly'))
);

-- Example shard map
INSERT INTO shards VALUES
    (1, 1, 1000000, 'db1.example.com:5432', 'active'),
    (2, 1000000, 2000000, 'db2.example.com:5432', 'active'),
    (3, 2000000, 3000000, 'db3.example.com:5432', 'draining');

Each write or read passes through a routing layer. A client library using libpq in PostgreSQL may use a shard_map table and a SELECT to locate the shard for a key:

def get_user(user_id):
    cursor.execute("""
        SELECT node_address, start_key, end_key
        FROM shards
        WHERE start_key <= %s AND end_key > %s
        ORDER BY start_key DESC
        LIMIT 1
    """, (user_id, user_id))
    row = cursor.fetchone()
    if not row:
        raise KeyError(f"No shard for user ID {user_id}")
    node, start, end = row
    # Connect to node and execute query
    with psycopg2.connect(f"host={node} user=app dbname=users") as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            return cur.fetchone()

Common failure modes:

• A user_id falls between two shards but no shard covers the gap.
• A shard becomes unavailable; queries hang until timeout or failover.
• Metadata is stale; a node returns data from an outdated shard map.

Hash-Based Sharding

Use a hash function (e.g., hash(key) % N) to assign each row to one of N shards. This pattern spreads data evenly and avoids hotspots, especially when writes are random.

For example, a session store uses session_id as a key. The hash determines which shard holds the session.

$ echo "session_a1b2c3" | sha256sum
# Output: 3a1b4c5d... (64 hex chars)
$ echo "3a1b4c5d..." | xxd -r -p | od -An -tu4
# Output: 101020304
$ echo 101020304 | awk '{print $1 % 4}'
# Output: 0
# → Assign to shard 0

A client library may cache the hash-to-shard mapping in memory. When a new shard is added, the hash ring is updated, and existing keys are re-assigned.

Silent failure:

• A new shard is added but the routing layer does not refresh the mapping.
• A key is written to a shard, but the client does not recompute the hash, leading to a mismatch.

Key tradeoff:

• Hash-based sharding offers even distribution but requires load balancing during scale-out.
• Range-based sharding is better for range queries but suffers from hotspots and skew.

Consistency and Replication

Distributed databases replicate data across multiple nodes to ensure availability and durability. The pattern of replication and consistency is central to system behavior.

Leader-Based Replication

One node acts as the leader (or primary), accepting writes. Changes are propagated to followers (or replicas) via a log.

Writes are accepted at the leader and asynchronously replicated to followers. The leader maintains a write-ahead log (WAL), and followers consume entries via a replication stream.

A client connects to the leader and executes:

-- Write to leader
BEGIN;
INSERT INTO orders (id, user_id, amount) VALUES (12345, 9876, 150.00);
COMMIT;

The leader writes the transaction to WAL and sends a LOG_ENTRY message to each follower:

{
  "term": 5,
  "index": 123,
  "type": "INSERT",
  "table": "orders",
  "data": {
    "id": 12345,
    "user_id": 9876,
    "amount": 150.00
  }
}

Followers apply entries in order. The leader can send a ACK for each entry. Clients may wait for acknowledgment before considering a write durable.

Synchronous vs Asynchronous Replication

The timing of replication affects consistency and availability.

• Asynchronous replication: Leader returns after writing to WAL. Followers apply entries in the background.
• Pros: Low latency, high throughput.
• Cons: Possible data loss if leader fails before followers commit.

• Synchronous replication: Leader waits for at least one follower to acknowledge the log entry before returning.

-- Client waits for replication acknowledgment
SET synchronous_commit = on;
SET commit_delay = 100; -- Wait 100ms for sync

The leader must wait for a REPLICATION_ACK from a follower before committing. If the follower fails before acknowledging, the leader may retry or promote a new leader.

Critical edge:

• A write is acknowledged but the follower fails before syncing the next entry.
• The leader crashes just after acknowledging; the write is durable but not yet committed.

Cross-Node Transactions

When data spans multiple shards, transactions must span multiple nodes.

Two-Phase Commit (2PC)

A protocol to coordinate a transaction across multiple databases.

1. Prepare phase: Coordinator sends PREPARE to all participants.
2. Commit phase: If all participants vote YES, coordinator sends COMMIT to all.

The coordinator maintains a transaction_log:

CREATE TABLE transaction_log (
    tx_id UUID PRIMARY KEY,
    status TEXT CHECK (status IN ('prepared', 'committed', 'aborted')),
    coordinator_node TEXT,
    participants JSONB
);

A transaction to transfer funds between two users, each in different shards:

-- Phase 1: Prepare
BEGIN;
-- On source shard
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
INSERT INTO transaction_log (tx_id, status, coordinator_node, participants)
VALUES ('a1b2c3', 'prepared', 'shard-1', '{"shard-1": true, "shard-2": false}');

-- On target shard
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;
INSERT INTO transaction_log (tx_id, status, coordinator_node, participants)
VALUES ('a1b2c3', 'prepared', 'shard-1', '{"shard-1": true, "shard-2": true}');

-- Send PREPARE to both
-- Wait for ACKs
COMMIT; -- Coordinator commits

If a participant fails during prepare, the coordinator sends ROLLBACK and retries.

Failure modes:

• A participant fails after sending PREPARE but before receiving COMMIT.
• The coordinator fails after sending COMMIT but before receiving acknowledgments.
• A participant receives COMMIT but fails before writing the result.

Distributed Transaction with Spanning Coordinator

For high-throughput systems, a dedicated coordinator service manages transactions across shards.

The coordinator stores transaction state and orchestrates phases. It may use a message queue (e.g., Kafka) to communicate with shard nodes.

{
  "tx_id": "d3e4f5",
  "phase": "prepare",
  "participants": [
    {
      "node": "shard-1",
      "state": "preparing",
      "last_heartbeat": "2025-04-05T12:00:00Z"
    },
    {
      "node": "shard-2",
      "state": "prepared",
      "last_heartbeat": "2025-04-05T12:00:01Z"
    }
  ],
  "start_time": "2025-04-05T11:59:58Z"
}

The coordinator polls each participant, timeouts after 5 seconds. If no heartbeat, it retries.

Critical detail:

• A participant sends PREPARE but the coordinator fails before recording it.
• The coordinator must handle idempotent operations: sending COMMIT multiple times is safe.

Handling Failures and Dynamic Reconfiguration

A distributed database must survive node failures, network partitions, and scale.

Failover and Leader Election

When the leader fails, followers elect a new leader. A common pattern uses a shared log with a leader_election table.

Each follower logs its heartbeat to a ping table:

CREATE TABLE ping (
    node_id TEXT PRIMARY KEY,
    last_heartbeat TIMESTAMPTZ,
    status TEXT CHECK (status IN ('alive', 'dead'))
);

A leader election process runs every 10 seconds:

# Every 10s: check for dead nodes
$ psql -c "
  UPDATE ping SET status = 'alive'
  WHERE node_id = 'shard-1' AND last_heartbeat > now() - interval '15s';

$ psql -c "
  UPDATE ping SET status = 'dead'
  WHERE node_id IN (
    SELECT node_id FROM ping
    WHERE status = 'alive' AND last_heartbeat < now() - interval '15s'
  );
"

The node with the highest last_heartbeat becomes leader. If multiple nodes are alive, the one with the highest node_id (lexicographically) wins.

A new leader writes to the leader_election table:

INSERT INTO leader_election (node_id, term, start_time)
VALUES ('shard-3', 6, now())
ON CONFLICT (node_id) DO UPDATE
SET term = EXCLUDED.term, start_time = EXCLUDED.start_time;

Silent failure:

• A node writes a heartbeat but the UPDATE fails due to network partition.
• Two nodes vote for leader but the winner’s write is delayed.

Dynamic Scaling and Rebalancing

When nodes are added or removed, data must be rebalanced across the cluster.

A rebalancing process:

1. Identifies under-utilized or overloaded nodes.
2. Moves data from one node to another.
3. Updates the shard map.

# Rebalance shards: move shard 2 from db2 to db4
$ psql -c "
  UPDATE shards
  SET node_address = 'db4.example.com:5432', status = 'active'
  WHERE shard_id = 2;
"

# Notify clients of new shard location
$ curl -X POST http://metadata-service:8080/shards/updated \
  -H 'Content-Type: application/json' \
  -d '{
    "shard_id": 2,
    "new_node": "db4.example.com:5432",
    "timestamp": "2025-04-05T12:10:00Z"
  }'

Clients may cache the shard map and refresh it every 10 seconds. A client library may use a shard_map_version column to detect changes.

CREATE TABLE shard_map (
    version INT PRIMARY KEY,
    last_updated TIMESTAMPTZ,
    shards JSONB
);

When a client sees a higher version, it reloads the map.

Critical failure:

• A shard is moved, but the metadata is updated before the data is copied.
• Clients read stale data from the old node.

Cross-Datacenter Replication

For low-latency access across regions, data is replicated across data centers.

A pattern: active-active replication, where each data center accepts writes and replicates changes to others.

Each data center runs a leader, and changes are propagated via a message queue or direct stream.

{
  "event": "write",
  "table": "orders",
  "data": {
    "id": 12345,
    "user_id": 9876,
    "amount": 150.00
  },
  "source_dc": "us-west",
  "timestamp": "2025-04-05T12:00:00Z"
}

Each data center consumes messages and applies changes. Conflicts are resolved using timestamps or version vectors.

Failure mode:

• Two data centers write to the same row during a network partition.
• A client reads from DC A, but a write from DC B is not yet replicated.
• The system must resolve conflicts using a conflict_resolver function.

This page defines the essential patterns of distributed databases, grounded in real mechanics, not vendor-specific abstractions. Each pattern addresses a core engineering challenge: how to distribute, replicate, coordinate, and recover data across a network of independent nodes.

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.