Databases at Scale#

A single database server eventually becomes a bottleneck — whether for read throughput, write throughput, storage capacity, or availability. Scaling databases requires understanding replication, sharding, and the trade-offs each introduces.

Replication#

Replication maintains copies of data across multiple database nodes for availability, durability, and read scaling.

Synchronous vs Asynchronous Replication#

sequenceDiagram
    participant Client
    participant Primary
    participant Replica1
    participant Replica2

    Note over Client,Replica2: Synchronous Replication
    Client->>Primary: INSERT INTO orders...
    Primary->>Replica1: Replicate (wait for ACK)
    Replica1-->>Primary: ACK
    Primary-->>Client: Commit confirmed

    Note over Client,Replica2: Asynchronous Replication
    Client->>Primary: INSERT INTO orders...
    Primary-->>Client: Commit confirmed (immediate)
    Primary->>Replica2: Replicate (eventual)

Semi-Synchronous#

A hybrid: the primary waits for at least one replica to acknowledge before confirming to the client. Other replicas receive updates asynchronously.

Write → Primary → Wait for 1 replica ACK → Confirm to client
                → Async replicate to remaining replicas

This balances durability (no data loss if primary + one replica survive) with write performance.

Primary-Replica (Master-Slave)#

flowchart TD
    App[Application] -->|Writes| Primary[(Primary)]
    App -->|Reads| Replica1[(Replica 1)]
    App -->|Reads| Replica2[(Replica 2)]
    App -->|Reads| Replica3[(Replica 3)]
    Primary -->|Replication Stream| Replica1
    Primary -->|Replication Stream| Replica2
    Primary -->|Replication Stream| Replica3

Read-after-write consistency problem: User writes to primary, immediately reads from replica that hasn’t received the update yet.

Solutions:

1. Route reads to primary for recently-written data (time window)
2. Route reads to primary for the user who wrote (session-level)
3. Client tracks write timestamp; replica must be at least that fresh

Multi-Primary (Master-Master)#

flowchart LR
    App1[App Region A] -->|Reads + Writes| Primary1[(Primary A)]
    App2[App Region B] -->|Reads + Writes| Primary2[(Primary B)]
    Primary1 <-->|Bidirectional Replication| Primary2

Conflict resolution strategies:

• Last-write-wins (LWW) — timestamp-based, simple but can lose data
• Application-level — custom merge logic per domain
• CRDTs — conflict-free replicated data types (counters, sets)
• Avoid conflicts — partition writes by key range or region

Read Replicas#

Dedicated replicas optimized for read queries, offloading the primary.

Routing Logic#

class DatabaseRouter:
    def __init__(self, primary, replicas):
        self.primary = primary
        self.replicas = replicas
        self._index = 0
    
    def get_connection(self, operation: str):
        if operation == "write":
            return self.primary
        # Round-robin across read replicas
        replica = self.replicas[self._index % len(self.replicas)]
        self._index += 1
        return replica
    
    def get_fresh_read(self):
        """For read-after-write consistency."""
        return self.primary

Read Replica Considerations#

Connection Pooling#

Database connections are expensive (TCP handshake, TLS, authentication, memory allocation). Pooling reuses connections across requests.

Without Pooling#

Request 1: Open connection → Query → Close connection  (50ms overhead)
Request 2: Open connection → Query → Close connection  (50ms overhead)
Request 3: Open connection → Query → Close connection  (50ms overhead)

With Pooling#

Startup: Create pool of 20 connections

Request 1: Borrow connection → Query → Return to pool  (< 1ms overhead)
Request 2: Borrow connection → Query → Return to pool  (< 1ms overhead)
Request 3: Borrow connection → Query → Return to pool  (< 1ms overhead)

Pool Configuration#

Connection Pool Architectures#

PgBouncer pooling modes:

• Session — connection held for entire client session
• Transaction — connection returned after each transaction (recommended)
• Statement — connection returned after each statement (most aggressive)

Sharding (Horizontal Partitioning)#

Sharding splits data across multiple database instances, each holding a subset. Unlike replication (copies of all data), sharding distributes different data to different nodes.

Data Partitioning vs Sharding#

Partitioning improves query performance (scan less data). Sharding enables horizontal scalability (each shard handles a fraction of total traffic).

Hash-Based Sharding#

def get_shard(user_id: str, num_shards: int) -> int:
    return hash(user_id) % num_shards

flowchart TD
    Router[Shard Router] -->|hash % 4 = 0| S0[(Shard 0)]
    Router -->|hash % 4 = 1| S1[(Shard 1)]
    Router -->|hash % 4 = 2| S2[(Shard 2)]
    Router -->|hash % 4 = 3| S3[(Shard 3)]

Range-Based Sharding#

Assign contiguous ranges of the shard key to each shard.

Shard A: user_id 1 – 1,000,000
Shard B: user_id 1,000,001 – 2,000,000
Shard C: user_id 2,000,001 – 3,000,000

Directory-Based Sharding#

A lookup table maps each key (or key range) to its shard.

Lookup Table:
  user_123 → Shard B
  user_456 → Shard A
  user_789 → Shard C

Consistent Hashing with Virtual Nodes#

Solves the key redistribution problem when shards are added or removed.

graph TD
    subgraph "Hash Ring"
        direction LR
        VN1[Shard A - VNode 1<br/>pos: 500]
        VN2[Shard B - VNode 1<br/>pos: 1500]
        VN3[Shard A - VNode 2<br/>pos: 2800]
        VN4[Shard C - VNode 1<br/>pos: 4000]
        VN5[Shard B - VNode 2<br/>pos: 5500]
        VN6[Shard C - VNode 2<br/>pos: 7200]
    end

How it works:

1. Each physical shard maps to N virtual nodes (typically 100–200) on the hash ring
2. A key hashes to a position and routes to the next virtual node clockwise
3. When a shard is added, it claims virtual node positions, taking ~1/total_vnodes of keys from neighbors
4. When removed, its virtual nodes’ key ranges are absorbed by the next node clockwise

Properties:

• Adding a shard moves only 1/N of keys (vs. all keys with modular hashing)
• Virtual nodes ensure even load even with few physical shards
• Used by DynamoDB, Cassandra, and most distributed databases

Cross-Shard Queries#

The biggest challenge of sharding: queries that span multiple shards.

The Problem#

-- Easy: single-shard query (shard key in WHERE clause)
SELECT * FROM orders WHERE user_id = 'user_123';

-- Hard: cross-shard query (no shard key filter)
SELECT * FROM orders WHERE total > 1000 ORDER BY created_at DESC LIMIT 10;

Approaches#

Shard Key Selection#

The shard key is the most critical decision in a sharded architecture:

NewSQL Databases#

NewSQL databases attempt to provide the scalability of NoSQL with the ACID guarantees and SQL interface of traditional relational databases.

How They Work#

flowchart TD
    Client[SQL Client] --> Router[Query Router / Coordinator]
    Router --> Shard1[(Shard 1<br/>Raft Group)]
    Router --> Shard2[(Shard 2<br/>Raft Group)]
    Router --> Shard3[(Shard 3<br/>Raft Group)]
    
    subgraph "Each Shard = Raft Group"
        Leader[Leader] --> F1[Follower 1]
        Leader --> F2[Follower 2]
    end

Key innovations:

• Automatic sharding — the database handles data distribution transparently
• Distributed transactions — Two-phase commit or equivalent across shards
• Consensus replication — Raft/Paxos within each shard for strong consistency
• SQL interface — Standard SQL, joins, indexes work transparently across shards

When to Choose NewSQL#

Scaling Decision Framework#

flowchart TD
    Start[Database Under Pressure] --> ReadHeavy{Read-heavy<br/>or Write-heavy?}
    
    ReadHeavy -->|Read-heavy| Cache[Add Caching Layer]
    Cache --> StillSlow{Still slow?}
    StillSlow -->|Yes| ReadReplicas[Add Read Replicas]
    ReadReplicas --> StillMore{Still need more?}
    StillMore -->|Yes| Shard[Shard the Database]
    
    ReadHeavy -->|Write-heavy| Vertical[Vertical Scaling First]
    Vertical --> StillHot{Still hitting limits?}
    StillHot -->|Yes| WriteSharding[Shard by Write Key]
    
    ReadHeavy -->|Both| Combined[Read Replicas + Sharding]

Scaling Progression#

Key Takeaways#

1. Replication scales reads, not writes — every replica receives every write. For write scaling, you need sharding or a distributed database.

2. Async replication trades consistency for performance — accept replication lag as a design constraint. Build your application to tolerate stale reads where possible.

3. Connection pooling is almost always free performance — the overhead of connection creation dominates in most applications. Use PgBouncer or application-level pooling.

4. Shard key selection is irreversible — choose a key that is high-cardinality, uniformly distributed, and present in most queries. Changing it later requires a full data migration.

5. Consistent hashing minimizes redistribution — with virtual nodes, adding a shard moves only 1/N of data. This is why every major distributed database uses it.

6. Cross-shard queries are the tax of sharding — design your data model to co-locate related data on the same shard. Avoid joins that span shards.

7. NewSQL is not magic — it pushes complexity into the database layer rather than eliminating it. Network latency, CAP theorem, and distributed transaction costs still apply. Use when you genuinely need horizontal scale with ACID.

8. Scale progressively — don’t shard on day one. Optimize queries → add caching → add replicas → vertical scaling → shard. Each step buys time before needing the next.