Scalability & Performance

Database Scaling Patterns: Replicas, Shards, Pools, and Query Tuning

Master database scaling with read replicas, sharding, materialized views, denormalization, connection pooling, and query optimization techniques.

read replicasshardingmaterialized viewsconnection poolingquery optimization

Start With an Order History Page That Times Out

Your SaaS platform has an order history page. It shows customers their past orders with filtering, sorting, and search. For the first three years, it loads in 200ms. The orders table has 10 million rows.

By year four, the table has 100 million rows. The order history page now takes 12 seconds to load. Customers are complaining. The support team has a running bet on how many seconds it will be by Friday.

The team adds more application servers. No improvement. They add Redis caching. The first page load is still slow because the cache is cold for every new query combination. They double the database RAM. It helps for a week. Then the table grows another 5 million rows and the page is slow again.

The problem is not server capacity. It is not cache size. It is a database that was designed for 10 million rows trying to serve 100 million rows with the same query patterns, indexes, and schema. The database needs to be redesigned for scale, not just provisioned with more resources.

Mental model: Database scaling is not about making a single database faster. It is about matching the database architecture to the data size and access pattern. A database that works at 10 million rows can fail at 100 million rows even on the same hardware — because the query planner makes different decisions at different scales.


Why Database Scaling Is Hard

Application servers scale horizontally with relative ease because they can be stateless. You add more instances behind a load balancer and traffic distributes evenly.

Databases cannot do this. They own durable state. They must maintain consistency across indexes, transactions, locks, and replication. You cannot just put two databases behind a load balancer and expect them to stay in sync without careful engineering.

⚠️

Important distinction: Adding application instances solves capacity. It does not solve database bottlenecks. If the database is the slowest component, more app instances will queue on database connections and make the problem worse. Scale the database first, then scale the application.


Why Not Just Use a Bigger Database Instance?

Beginners often ask: "Why not just move to a larger RDS instance with more CPU and RAM?"

Vertical scaling works — up to a point. A bigger instance gives the database more memory for caching, more CPU for query processing, and more IOPS for disk access. But it does not change the fundamental architecture.

BottleneckBigger Instance Helps?Limitation
Query is slow (missing index)No — same query plan on bigger hardwareStill scans too many rows
Too many readsYes — more memory caches more dataEventually hits instance size limit
Too many writesPartially — more CPU helps, but single-writer bottleneck remainsSingle primary handles all writes
Table too large for indexesNo — index size grows with data, not instance sizeB-tree depth increases, more disk I/O per query
Connection limitYes — bigger instance allows more connectionsEach connection still consumes memory
Storage capacityYes — bigger volumesVolume size limit per instance type
Geo-distributionNo — all traffic goes to one regionCross-region latency is a physics problem

The correct approach is a decision path: optimize queries → add caching → scale reads with replicas → scale writes with sharding. Jumping straight to a bigger instance skips cheaper, more effective solutions.


Step 1: Diagnose the Bottleneck Before Scaling

The most common mistake in database scaling is applying a solution before understanding the problem. Every database bottleneck looks like "the database is slow." But the root cause can be very different.

What to Check First

CheckToolWhat It Reveals
Slow query logpg_stat_statements, slow_query_logWhich queries consume the most time
Query planEXPLAIN ANALYZEIndex scan vs sequential scan, join strategy
Connection countpg_stat_activity, SHOW CONNECTIONSAre you running out of connections?
Lock contentionpg_locks, SHOW PROCESSLISTAre queries waiting on locks?
Cache hit ratepg_statio_user_tablesAre indexes and data cached in memory?
IO utilizationiostat, CloudWatchIs the disk saturated?
Replication lagpg_stat_replicationAre replicas falling behind?

Step 2: Query Optimization and Indexing

Before adding replicas or shards, make sure the existing database is running efficient queries. A single missing index can cause 100x more I/O than necessary.

Query Plan Analysis

sql
-- Before: 12 seconds, sequential scan on 100M rows
EXPLAIN ANALYZE
SELECT * FROM orders
WHERE customer_id = 456
AND status = 'completed'
ORDER BY created_at DESC
LIMIT 20;

-- Seq Scan on orders (cost=0.00..2345678.00 rows=1 width=120)
-- Filter: ((customer_id = 456) AND (status = 'completed'::text))
-- Rows Removed by Filter: 99,999,999
-- Planning Time: 0.5 ms
-- Execution Time: 12,345 ms

-- The database scanned 100 million rows to find 20 matching rows.
sql
-- After adding composite index: 3ms
CREATE INDEX idx_orders_customer_status_created
ON orders (customer_id, status, created_at DESC);

EXPLAIN ANALYZE
SELECT id, total, status, created_at
FROM orders
WHERE customer_id = 456
AND status = 'completed'
ORDER BY created_at DESC
LIMIT 20;

-- Index Scan using idx_orders_customer_status_created (cost=0.56..12.45 rows=20 width=48)
-- Index Cond: ((customer_id = 456) AND (status = 'completed'::text))
-- Planning Time: 0.3 ms
-- Execution Time: 2.8 ms

-- The database scanned exactly 20 rows. 12 seconds → 3 milliseconds.

Index Selection Guide

Query PatternIndex TypeExample
Equality on one columnSingle-column B-treeWHERE customer_id = ?
Equality + rangeComposite B-tree (equality column first)WHERE status = ? AND created_at > ?
SortingB-tree on sort column, matching sort orderORDER BY created_at DESC
Text searchGIN or pg_trgmWHERE name ILIKE '%search%'
JSONB queriesGIN on JSONB columnWHERE metadata @> '{"tier": "premium"}'
Partial table subsetPartial indexWHERE status = 'active' (index only active rows)
Covering all selected columnsCovering index (INCLUDE)CREATE INDEX ... INCLUDE (total, status)

The Index Trade-off

BenefitCostWhen It Bites
Faster readsSlower writes (indexes must be updated)Write-heavy tables with too many indexes
Faster sortingMore storage spaceLarge tables with multiple composite indexes
Better joinsMore maintenance (VACUUM, ANALYZE)Frequently updated columns
Unique constraintsWrite contention on unique checkHigh-concurrency insert on unique column

Practical advice: Add indexes based on actual query patterns from production, not theoretical ones. The slow query log tells you exactly which queries need indexing. Do not add indexes pre-emptively — every index slows writes and consumes storage.


Step 3: Connection Pooling

Databases cannot handle unlimited connections. Each connection consumes memory (2-10MB in PostgreSQL), CPU for context switching, and database resources for maintaining session state.

How Connections Become a Bottleneck

txt
Scenario: A busy application with 50 API servers

Without connection pooling:
  50 servers × 10 connections each = 500 database connections
  PostgreSQL default max_connections = 100
  → 400 connection attempts are queued or rejected
  → Applications throw timeout errors
  → Status: 502 Bad Gateway

With connection pooling (PgBouncer):
  50 servers → PgBouncer (50 connections) → Database (50 connections)
  Wait: only 50 connections to the database instead of 500
  → PgBouncer multiplexes application connections to database connections
  → Applications never wait for database connections
  → Status: healthy

Pool Size Calculation

txt
Formula:
  pool_size = (max_expected_requests_per_second × avg_query_duration_seconds)
              + spare_capacity

Example:
  Peak requests per second: 1000
  Average query duration: 50ms (0.05s)
  Required pool: 1000 × 0.05 = 50 connections
  With 2x spare: 100 connections

  A PostgreSQL instance with 100 connections and 2GB shared_buffers
  can handle this comfortably. Without pooling, 50 app servers × 10 connections
  = 500 connections would overwhelm it.

Common Pooling Problems

ProblemRoot CauseFix
Too many app instances create too many connectionsEach instance opens its own pool, multiplied across instancesUse centralized pooler (PgBouncer, ProxySQL, RDS Proxy)
Long transactions hold connectionsTransaction spans multiple queries with application logic in betweenKeep transactions short — open, query, commit, close
Pool too smallRequests queue in the application, latency increasesIncrease pool size based on query duration × concurrency
Pool too largeDatabase thrashes from too many concurrent queriesDecrease pool size — fewer connections often improves throughput
Serverless burst opens many connectionsLambda/Cloud Run creates many instances simultaneouslyUse a data API (Neon, PlanetScale) or pooler with serverless support

Step 4: Read Replicas

When the database bottleneck is read volume (the most common case), read replicas are the simplest and most effective solution.

Benefits and Trade-offs

BenefitTrade-offMitigation
Scales read throughput linearlyReplication lag (ms to seconds)Monitor lag, alert if > threshold
Reduces primary loadRead-after-write inconsistencyRead-your-writes strategies
Improves regional latency (deploy replicas in user regions)Cross-region replication costOnly replicate necessary tables
Isolates analytics/reporting from production trafficReplica may have different hardware specDedicated replica for analytics

Read-After-Write Consistency Strategies

When a user writes data and then immediately reads it, they may see stale data on a replica that has not caught up.

StrategyHow It WorksComplexityConsistency
Read your writes from primaryRoute reads from the same user to primary for N seconds after their last writeLowStrong
Session consistency tokenPrimary returns a token representing current LSN; replica serves reads only after it reaches that LSNMediumStrong
Sticky read routingSame user reads the same replica for the duration of their sessionLowDepends on replica lag
UI delay with optimistic UIShow the written data immediately in the UI (client-side), serve reads from replica asynchronouslyMediumEventual (but user sees their write)
⚠️

Replica lag is a product behavior, not just a database detail: If a user updates their profile and the next page load shows the old data, that is a product bug, not a database feature. Design the read path intentionally. Choose a consistency strategy and test it with real user flows.


Step 5: Materialized Views and Denormalization

Sometimes the best scaling move is to do less work at read time. Precompute expensive queries so the database returns pre-calculated results instead of computing them on every request.

Materialized Views

A materialized view stores the result of a query physically. It is refreshed on a schedule or on demand.

sql
-- Without materialized view: complex JOIN scanned every time
-- Every page load runs this 4-second query

SELECT
  c.name,
  COUNT(o.id) as order_count,
  SUM(o.total) as total_spent,
  MAX(o.created_at) as last_order_date
FROM customers c
LEFT JOIN orders o ON o.customer_id = c.id
GROUP BY c.id;

-- With materialized view: precomputed, refreshed every 5 minutes
CREATE MATERIALIZED VIEW customer_order_summary AS
SELECT
  c.id,
  c.name,
  COUNT(o.id) as order_count,
  SUM(o.total) as total_spent,
  MAX(o.created_at) as last_order_date
FROM customers c
LEFT JOIN orders o ON o.customer_id = c.id
GROUP BY c.id;

CREATE UNIQUE INDEX ON customer_order_summary (id);

-- Query: 4ms instead of 4 seconds
SELECT * FROM customer_order_summary WHERE id = 456;

-- Refresh:
REFRESH MATERIALIZED VIEW CONCURRENTLY customer_order_summary;

Read Model Design

PatternBest ForHow It WorksTrade-off
Materialized viewExpensive aggregate queriesSQL-based precomputation, periodically refreshedStale data between refreshes
Denormalized tableFast read model for a specific query patternFlatten related data into one table, update on writesWrite amplification — every write must update multiple copies
Search index (Elasticsearch)Text search, faceted search, complex filteringIndex documents from database, serve reads from search engineEventual consistency, operational complexity
Analytics store (ClickHouse, Redshift)OLAP and reporting queriesColumnar storage, optimized for aggregationsSeparate pipeline to load data

Denormalization Trade-off

txt
Normalized schema:
  orders table: id, customer_id, total, status, created_at
  order_items table: id, order_id, product_id, quantity, price

  Query to show order with items (needs JOIN): 50ms
  Write an order (one INSERT into orders + N inserts into order_items): 5ms

Denormalized schema:
  orders_denormalized table: id, customer_id, total, status, created_at,
    items_json (JSON or array of items)

  Query to show order with items (no JOIN): 2ms
  Write an order (one INSERT including JSON): 6ms (slightly slower write)

When reads >> writes: denormalization saves 48ms per read at the cost of 1ms per write.
If you read orders 100x more than you write them, the trade-off is obvious.

Step 6: Sharding

Sharding splits data across multiple databases. Each shard owns a subset of the data and operates independently. This is the most powerful database scaling technique and the most complex.

Sharding Strategies

StrategyHow It WorksDistributionBest ForWorst For
Hash-basedhash(shard_key) % num_shardsEvenGeneral purposeAdds/removes shards (changes mapping)
Range-basedKey ranges per shard (IDs 0-1M in shard 1)May be unevenSequential accessHot spots on latest range
Directory-basedLookup service maps key to shardFlexibleDynamic shard managementLookup latency, directory is SPOF
Geo-basedData lives in user's regionGrouped by locationMulti-region appsCross-region queries
Tenant-basedEach tenant maps to a shardPer-tenant isolationSaaS multi-tenantLarge tenants become hot

Choosing a Shard Key

A good shard key must:

  1. Appear in most queries. If queries do not include the shard key, the router must scatter-gather across all shards.
  2. Spread load evenly. If 90% of queries use 10% of the keys, sharding does not help.
  3. Avoid hot partitions. A celebrity user on a social media platform can overwhelm a single shard.
  4. Keep transactions local. Cross-shard transactions are slow and complex.
  5. Have stable ownership. A shard key that changes (e.g., a user's region) makes rebalancing painful.

Scatter-Gather: What Happens Without the Shard Key

A scatter-gather query sends the same query to every shard, waits for all responses, and merges results. If any shard is slow, the whole query is slow. The latency of a scatter-gather query is the latency of the slowest shard, not the average.


Step 7: Rebalancing

Shards grow unevenly over time. Data distribution changes. You need to add capacity. Rebalancing moves data between shards.

Rebalancing Approaches

ApproachData MovementComplexityDowntime
Consistent hashingMinimal (only 1/N of data)MediumNone
Virtual shardsMinimal (remap logical shard to new physical node)MediumNone
Live migration with dual writesAll data must be copiedHighNone (online)
Offline migrationAll data must be copiedLowSignificant downtime
Re-shard with new hash functionAll data must be re-distributedVery highMaintenance window

Consistent Hashing

Consistent hashing maps each shard key to a position on a hash ring. Each physical shard owns a range of the ring. When a shard is added, only the keys in the affected range need to be moved — not all keys.

txt
Naive modulo: hash(key) % 4
  Adding a 5th shard changes hash(key) % 5
  → 80% of keys map to a different shard
  → 80% of data must be moved

Consistent hashing: hash(key) on a ring with 4 shards
  Adding a 5th shard
  → Only keys in the new shard's range move
  → ~20% of data must be moved

Practical advice: Before you shard, exhaust every other option. Index optimization, caching, read replicas, materialized views, and connection pooling can handle the vast majority of database scaling needs. Sharding is for when you need to scale writes — not reads, not connections, not query performance. Only shard when writes to a single database are the bottleneck.


Common Failure Stories

The Missing Index That Cost $50,000

A team spends $50,000 on a larger database instance because the order history page is slow. The new instance improves performance for two weeks. The team celebrates. Then traffic grows and the page is slow again. A consultant runs EXPLAIN ANALYZE and finds a missing composite index. Adding the index takes 5 minutes. Query time drops from 8 seconds to 30ms. The $50,000 was wasted.

The fix: always run EXPLAIN ANALYZE before spending money on infrastructure. The slow query log tells you exactly what is wrong. Indexing is cheaper than any hardware upgrade.

The Replica Lag That Caused a Support Nightmare

A user updates their email address. The next page load shows the old email because the read replica has 3 seconds of lag. The user updates the email again. Now the primary has two different update requests. The replica catches up and shows the old email again. The user updates a third time. The support ticket arrives: "Your system keeps changing my email back."

The fix: implement read-your-writes consistency. After a write, route the user's reads to the primary for at least 5 seconds. Or use a session token that ensures the replica has caught up before serving the read.

The Connection Storm

A microservice deployment launches 30 new instances simultaneously. Each instance opens a connection pool of 10 connections. That is 300 new connections to the database in under 30 seconds. The database reaches max_connections and starts rejecting. All 30 instances fail health checks. The deployment rolls back. The database remains near max_connections for 10 minutes.

The fix: use a centralized connection pooler (PgBouncer, RDS Proxy). Set max_connections with a safety margin. Implement a connection limit in the application that prevents pool explosion during rapid scaling.

The Shard Key That Did Not Spread Evenly

A SaaS platform shards by company ID using hash-based sharding. Company "Acme Corp" has 500,000 users and generates 60% of all traffic. The shard containing Acme's data is overloaded while other shards are idle. The system is not scalable — it is just relocating the bottleneck.

The fix: shard by user ID, not company ID, for systems with varying tenant sizes. If company-level sharding is required, split large tenants across multiple shards using a sub-shard key (e.g., company_id + user_id % N).

The Materialized View That Was Never Refreshed

A team creates a materialized view for the analytics dashboard. It works perfectly on day one — 4ms queries instead of 4 seconds. Nobody sets up a refresh schedule. A week later, the dashboard shows stale data. The product team makes decisions based on it for three days before someone notices the numbers are wrong.

The fix: always set up a refresh schedule when creating a materialized view. Define the refresh interval based on how stale the data can be. Monitor refresh failures with alerts. For critical data, use REFRESH MATERIALIZED VIEW CONCURRENTLY to avoid blocking reads.


Evaluating Database Scaling

You cannot know if your database scaling strategy is working without measuring the right signals.

What to Monitor

SignalWhat It RevealsTarget
Query latency (P50/P95/P99)Are queries fast enough?P50 < 10ms, P95 < 100ms
Slow query count per minuteAre new slow queries appearing?0 sustained
Connection utilizationAre you running out of connections?< 80% of max
Replication lagAre replicas keeping up?< 1 second
Cache hit rateAre indexes and data in memory?> 99% for active dataset
IOPS and throughputIs disk the bottleneck?< 80% of provisioned IOPS
Lock wait timeIs contention a problem?< 1% of query time
Index size vs table sizeAre indexes bloated?Indexes < 2x table size

Load Testing Database Changes

Every database change should be load tested before production:

txt
Test: Adding a new composite index to orders table

Baseline (current state):
  Query: SELECT * FROM orders WHERE customer_id = ? AND status = 'completed'
  P50: 450ms, P99: 3200ms
  Write throughput: 500 writes/second

After index:
  Query: SELECT * FROM orders WHERE customer_id = ? AND status = 'completed'
  P50: 3ms, P99: 15ms
  Write throughput: 480 writes/second (slight decrease due to index maintenance)

Decision: Index is worth adding. 150x read improvement for 4% write reduction.

Debugging rule: When the database is slow, start at the query level. Run EXPLAIN ANALYZE on the slowest query from pg_stat_statements. If the query plan shows a sequential scan on a large table, you need an index. If the query plan looks good but the database is still slow, check connections and locks. If those are fine, check IO and memory. The bottleneck is almost always at the query level, not the hardware level.


A Complete Database Scaling Path, End to End

Here is how the order history page scales from 10 million to 100 million orders:

Each stage addresses a specific bottleneck. Each stage is deployed only when the previous stage is exhausted. The result is a database architecture that can grow from millions to billions of rows without a complete rewrite.


What to Remember for Interviews

When explaining database scaling, tell the story in order:

  1. Measure before scaling: Run EXPLAIN ANALYZE. Check the slow query log. Know whether the bottleneck is queries, connections, locks, reads, or writes. The fix depends entirely on the bottleneck.
  2. Query optimization is the highest-impact, lowest-cost fix: A single missing index can turn a 12-second query into a 3ms query. Index for actual query patterns from production, not theoretical ones.
  3. Connection pooling protects the database: Each connection consumes memory and CPU. A centralized pooler allows many application instances to share a small number of database connections. Calculate pool size based on query duration × concurrency.
  4. Read replicas scale reads: The most common database bottleneck is read volume. Add read replicas. But replica lag is a product behavior — implement read-your-writes consistency.
  5. Materialized views precompute expensive reads: Before sharding, consider whether you can precompute the expensive queries. A materialized view can turn a 4-second aggregate query into a 4ms lookup.
  6. Sharding scales writes but adds significant complexity: Only shard when writes to a single database are the bottleneck. Choose a shard key that appears in most queries, spreads load evenly, and keeps transactions local. Use consistent hashing to make rebalancing practical.
  7. Rebalancing is a distributed systems project: Shards grow unevenly. Plan for rebalancing from day one. Use virtual shards or consistent hashing to minimize data movement.

Practice: Design a database scaling strategy for an order history system with 100 million orders, read-heavy traffic (1000 reads/second), occasional writes (50 writes/second), and enterprise tenant isolation. Start with the current architecture (single PostgreSQL instance). Walk through each stage: query optimization, caching, read replicas, materialized views, connection pooling, and sharding. Explain the trigger for each stage and the expected improvement.