system-design · intermediate

How to Scale a Database — The Progressive Scaling Ladder

The Central Question

Consider an enterprise analytics platform running on the DataLab platform (datalab.com) processing 100,000,000 queries per day:


A junior developer proposes an immediate solution: "We must immediately shard our database across 16 database servers!"

Attempting to jump straight to horizontal database sharding is one of the most expensive and dangerous mistakes in system design. Sharding introduces complex scatter-gather queries, cross-shard 2-Phase Commit (2PC) transactions, and complex operational resharding migrations.

In 90% of production scaling incidents, databases can scale by orders of magnitude using simpler, lower-risk architectural patterns.

This lesson answers one central question: What is the progressive 5-stage Database Scaling Ladder (Query Optimization $\rightarrow$ Vertical Scaling $\rightarrow$ Read Replica Offloading $\rightarrow$ Caching & Functional Partitioning $\rightarrow$ Horizontal Sharding), and how do engineers apply each stage to scale throughput while minimizing operational complexity?


The 5-Stage Database Scaling Ladder

Engineers resolve database capacity bottlenecks by advancing sequentially through five progressive engineering stages:

flowchart TD
  Stage1[Stage 1: Query & Index Optimization<br/>10x-100x Throughput Gain | Zero Infra Cost] --> Stage2[Stage 2: Vertical Hardware Scaling<br/>2x-5x Capacity Gain | Low Operational Risk]
  Stage2 --> Stage3[Stage 3: Read Replica Offloading<br/>5x-20x Read QPS Gain | Minor Replication Lag Trade-off]
  Stage3 --> Stage4[Stage 4: External Caching & Functional Partitioning<br/>10x Read Offload | Microservice Boundary Isolation]
  Stage4 --> Stage5[Stage 5: Horizontal Write Sharding<br/>Infinite Write Capacity | Maximum Operational Complexity]

Figure 1: The progressive 5-stage Database Scaling Ladder.


Stage 1: Query & Index Optimization (Zero Cost, Highest ROI)

Before spending budget on larger database instances or complex cluster topologies, engineers audit execution plans to eliminate wasteful sequential table scans.

flowchart LR
  subgraph Un-Indexed Query Execution
    Q1[SELECT * FROM orders WHERE user_id = 42] -->|Sequential Table Scan| Scan[Scans 50,000,000 Rows on Disk<br/>Execution Time: 4,500 ms]
  end

subgraph Index-Optimized Query Execution
Q2[SELECT * FROM orders WHERE user_id = 42] -->|B-Tree Index Seek| Seek[Traverses B-Tree Index<br/>Execution Time: 2 ms]
end

Figure 2: Comparing 50M-row sequential disk scans against 2ms B-Tree index seeks.


Stage 2: Vertical Scaling (Scale-Up)

When query optimization is complete and CPU or memory capacity is exhausted, the simplest operational move is Vertical Scaling (Scale-Up)—upgrading the underlying server instance with more CPU cores, RAM, and faster NVMe SSD storage.

flowchart LR
  subgraph Small Instance (db.m5.large)
    VM1[2 vCPUs<br/>8 GB RAM<br/>1,000 IOPS<br/>Max QPS: 1,500]
  end

subgraph Upgraded Instance (db.r6g.8xlarge)
VM2[32 vCPUs<br/>256 GB RAM<br/>32,000 IOPS<br/>Max QPS: 45,000]
end

Figure 3: Upgrading instance hardware vertically to expand CPU and RAM headroom.

Stage 3: Read Replica Offloading (Read-Write Splitting)

For read-heavy workloads (typically 80-90% reads, 10-20% writes), a single primary database becomes bottlenecked trying to process read queries alongside writes.

flowchart TD
  App[Application Servers] --> Router{Read-Write Split Router}
  
  Router -->|SQL Mutations: INSERT/UPDATE/DELETE| Primary[(Primary DB: Writes Only)]
  Router -->|SQL Queries: SELECT| LB[Read Replica Load Balancer]
  
  LB --> Replica1[(Read Replica 1)]
  LB --> Replica2[(Read Replica 2)]
  LB --> Replica3[(Read Replica 3)]
  
  Primary -.->|Async Replication| Replica1
  Primary -.->|Async Replication| Replica2
  Primary -.->|Async Replication| Replica3

Figure 4: Read-Write splitting routing writes to Primary and reads to Replica pool.

Read-Your-Own-Writes Consistency (Session Sticky Routing)

In asynchronous read-replica pools, a user who edits their user profile (writing to Primary DB) and immediately refreshes the page may hit Read Replica 1 before the WAL log replicates. The user sees their *old* un-edited profile data (**Stale Read Anomaly**). To solve this, database routers enforce **Read-Your-Own-Writes Consistency**:
  1. When a user submits an UPDATE mutation, the router sets a temporary 5-second session cookie or Redis token (user_last_write:timestamp).
  2. For the next 5 seconds, all subsequent SELECT queries from that specific user session are pinned directly to the Primary DB.
  3. Once the 5-second window elapses, the router switches the session reads back to the Read Replica pool, guaranteeing consistency without sacrificing replica offloading.

Connection Pooling Architecture (PgBouncer)

PostgreSQL spawns a separate dedicated OS process for every incoming database connection, consuming $\sim 10\text{ MB}$ of RAM per connection. Having 1,000 application pods open 50 connections each results in **50,000 PostgreSQL OS processes**, instantly crashing server memory. Systems deploy lightweight **Connection Poolers (PgBouncer)** between application pods and the database:

Connection Pool Formula

The optimal pool size for a PostgreSQL database is governed by the empirical PostgreSQL project formula:

$$N_{\text{connections}} = (\text{CPU Cores} \times 2) + \text{Spindle Count}$$

For a 16-core database server with NVMe SSD storage ($\text{Spindle Count} = 1$), the optimal connection pool size is:

$$N_{\text{connections}} = (16 \times 2) + 1 = 33 \text{ active connections}$$

Opening more than 33 connections increases CPU context-switching overhead, actually reducing overall query throughput!

Read Replica Health Probes & Load Balancing

When deploying a pool of Read Replicas, database load balancers (such as HAProxy or AWS Route53) continuously send health check probes (`SELECT 1`) to every replica instance. If Replica 2 suffers a hardware crash or exhibits excessive replication lag ($> 5\text{ seconds}$), the load balancer automatically ejects Replica 2 from the read pool. Incoming `SELECT` queries are redistributed across the remaining healthy replicas without application downtime.

Continuous Slow Query Log Auditing

In Stage 1 query optimization, operating teams configure PostgreSQL `log_min_duration_statement = 250ms` (logging any SQL query taking longer than $250\text{ms}$). Log collectors (e.g. Datadog or pganalyze) group slow queries by fingerprint, enabling engineers to run `EXPLAIN (ANALYZE, BUFFERS)` on the worst 1% of queries to identify missing indexes before high CPU incidents occur.

Index Maintenance & Vacuuming Overheads

While B-Tree indexes speed up `SELECT` reads from $O(N)$ to $O(\log N)$, every additional index slows down `INSERT`, `UPDATE`, and `DELETE` write mutations—because PostgreSQL must update every B-Tree index on disk during writes. SRE teams periodically audit unused indexes using `pg_stat_user_indexes` to drop redundant indexes and save write I/O.

Stage 4: Functional Partitioning & Caching

Instead of storing all tables (users, orders, payments, catalog, analytics) in a single monolithic database, Functional Partitioning splits databases along domain boundaries.

flowchart LR
  subgraph Monolithic Database
    Mono[(Users + Orders + Payments DB)]
  end

subgraph Functional Partitioning
UserDB[(User DB)]
OrderDB[(Order DB)]
PaymentDB[(Payment DB)]
end

Figure 5: Functional partitioning separating tables by domain microservices.


Stage 5: Horizontal Write Sharding

When a single primary database cannot handle write throughput even after vertical upgrades, systems implement Horizontal Sharding—partitioning rows across independent database servers based on a Shard Key.


Complete Worked Example: Production Go Read-Write Split Database Router

Let's inspect a complete Go implementation of a Read-Write Split Database Router for the DataLab platform (datalab.com).

package main

import (
"context"
"database/sql"
"fmt"
"math/rand"
"sync"
"sync/atomic"
)

type ReadWriteSplitDB struct {
primary sql.DB
replicas []
sql.DB
rrIndex uint64
}

func NewReadWriteSplitDB(primaryConn string, replicaConns []string) (*ReadWriteSplitDB, error) {
// Connect to Primary DB
pDB, err := sql.Open("postgres", primaryConn)
if err != nil {
return nil, fmt.Errorf("primary connection failed: %w", err)
}

// Connect to Read Replicas
replicas := make([]*sql.DB, 0, len(replicaConns))
for _, rConn := range replicaConns {
rDB, err := sql.Open("postgres", rConn)
if err == nil {
replicas = append(replicas, rDB)
}
}

return &ReadWriteSplitDB{
primary: pDB,
replicas: replicas,
}, nil
}

func (db *ReadWriteSplitDB) ExecWrite(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
// Writes ALWAYS route to Primary DB
fmt.Println("[WRITE ROUTER] Routing INSERT/UPDATE mutation to PRIMARY DB")
return db.primary.ExecContext(ctx, query, args...)
}

func (db ReadWriteSplitDB) QueryRead(ctx context.Context, query string, args ...interface{}) (sql.Rows, error) {
if len(db.replicas) == 0 {
return db.primary.QueryContext(ctx, query, args...)
}

// Round-Robin Selection across Read Replicas
idx := atomic.AddUint64(&db.rrIndex, 1) % uint64(len(db.replicas))
targetReplica := db.replicas[idx]

fmt.Printf("[READ ROUTER] Routing SELECT query to Read Replica #%d\n", idx)
return targetReplica.QueryContext(ctx, query, args...)
}


Failure Modes and Engineering Mitigations

Failure ScenarioRoot CauseSystem SymptomDetection SignalMitigation Strategy
1. Premature Sharding DisasterEngineering team implements horizontal sharding prematurely without indexing or replica pool optimization.Massive operational overhead, broken cross-table JOINs, and zero throughput gain.Developer productivity drops by 80%; operational outage rate surges.Follow the Progressive 5-Stage Scaling Ladder sequentially before sharding.
2. Replication Lag Read StaleAsynchronous replication lag causes a user to update profile details but see old details on immediate page refresh.Customers report profile updates or password changes failing to persist.Replication lag age metric $> 1000\text{ms}$ on Datadog dashboards.Implement Read-Your-Own-Writes Consistency (route reads to Primary for 5s after write).
3. Connection Pool ExhaustionWeb servers open unlimited SQL connections, causing PostgreSQL backend process memory exhaustion.Database returns FATAL: sorry, too many clients already errors.Active connection count reaches max limit; connection timeouts surge.Deploy external Connection Poolers (PgBouncer) and enforce strict pool size limits.
4. Replica Cascade CrashReplica 1 crashes; remaining 2 replicas crash sequentially under 150% overload from diverted reads.Domino crash of all read replicas following single node failure.Sequential node offline alerts across read replica clusters.Maintain N+1 Headroom Capacity across read replica pools ($< 66\%$ CPU during normal ops).

What You Should Remember

  1. Follow the 5-Stage Scaling Ladder: Optimize queries $\rightarrow$ Scale vertically $\rightarrow$ Offload reads to replicas $\rightarrow$ Cache & partition functionally $\rightarrow$ H-Shard.
  2. Read-Write splitting multiplies read capacity: Route INSERT/UPDATE mutations to Primary and SELECT reads across a pool of Read Replicas.
  3. Handle Replication Lag for Read-Your-Own-Writes: Route reads to the Primary for a few seconds following user writes to prevent stale reads.
  4. Use Connection Poolers (PgBouncer): Prevent database OS process memory exhaustion by capping total connection counts with lightweight proxies.
  5. Functional partitioning simplifies microservices: Split monolithic databases along domain boundaries (Users, Orders, Payments) before sharding tables.

Glossary of Terms

TermDefinition
Vertical Scaling (Scale-Up)Upgrading a single database server with more CPU cores, RAM, and faster NVMe SSD storage.
Horizontal Scaling (Scale-Out)Distributing a database workload across multiple independent physical database server nodes.
Read ReplicaA secondary read-only database instance that continuously receives WAL updates from the Primary.
Read-Write SplittingAn access pattern that routes write mutations to Primary and read queries to Read Replicas.
Functional PartitioningSplitting a monolithic database into separate database instances grouped by domain functionality.
Connection PoolingReusing a fixed pool of open database connections to eliminate per-request connection overhead.

Practice Scenario and Self-Assessment

Architecture Scenario

You are managing the database tier for an e-commerce platform (`store.datalab.com`): **Questions**:
  1. Evaluate whether to scale vertically or implement Read-Write Splitting with 3 Read Replicas.
  2. Formulate the read routing rule to guarantee Read-Your-Own-Writes consistency for user checkout edits.

Interactive Self-Assessment

Sharding destroys cross-table JOINs, requires complex 2-Phase Commit transactions, and adds massive operational overhead.

Sharding causes physical hardware electrical short-circuits in server power units.

Sharding revokes edge HTTPS TLS encryption certificates on load balancers.

Sharding cuts physical CPU hardware clock speeds in half.

It routes write mutations to the Primary DB and distributes read queries across multiple Read Replicas, scaling read throughput horizontally.

Read replicas automatically convert SQL table columns into flat text files.

Read-write splitting replaces public DNS nameservers with local hosts entries.

Read replicas format the underlying NVMe SSD disk drives on primary database servers.


What to Learn Next

Track: Data, Storage and Messaging

Previous: Hot Partition / Hot Key — When One Shard Takes All the Heat

Next: Rebalancing Shards Under Skewed Traffic

By Shubham Jain

All articles · Study paths

Shubham Jain · Learning Lab