Intermediate System design concept · Distributed Systems & Data · 40 mins read
Database Scaling Techniques
Scale a relational database past one primary: split it by domain, reshape data for reads, tune the queries, and reshard without downtime.
- Thinking in Systems
- How Systems Actually Work
- Handling Scale & Bottlenecks
- Distributed Systems & DataYou are here
- Real Systems & Production Thinking
Federation (Functional Partitioning)
Split one database into several databases by business domain so each one handles less traffic and can scale on its own.
Intuition
A single database often serves every part of the product: users, catalogue, orders, payments, analytics. Each domain competes for the same CPU, memory, buffer cache and write throughput, and a heavy query in one domain slows down all the others.
Federation is usually the first split a growing system makes, because it removes contention without changing how any single domain stores its data. It also lines up with service boundaries: each service can own its own database.
Mental Model
Draw the boundaries along domains that rarely need to be queried together. Users go to one database, products to another, orders to a third. Each database is smaller, handles fewer writes, keeps more of its working set in memory, and can be replicated, tuned or upgraded independently. Think of it like: Instead of one office handling HR, sales and accounting for the whole company, each department gets its own office. Each office is less crowded — but a question that needs all three now means visiting three offices.
Building Blocks
- Domain boundaries: Split along data that is read and written together. Users and their profiles belong together; orders and products are referenced across the boundary only by ID.
- Routing in the application: Each service or data-access layer knows which database owns which tables. There is no shared connection that can query everything.
- References by ID, not foreign keys: An order stores product_id, but the database can no longer enforce that the product exists. Integrity moves into application logic.
- Independent scaling: A read-heavy catalogue database can get replicas and caching while the write-heavy orders database gets a bigger primary or sharding.
Definitions
- Federation
-
Splitting a database into multiple databases by function or domain.
- Also called functional or vertical partitioning.
- Each database holds a complete domain, not a slice of rows.
- Cross-database join
-
Combining data that lives in two federated databases.
- The database cannot do it for you.
- The application queries both and merges, or keeps a denormalized copy.
- Database per service
-
A microservices pattern where each service exclusively owns its database.
- Federation is the data side of this pattern.
- Other services access the data only through the owning service's API.
Patterns
- Split by bounded context — When domains have clear ownership and few queries span them.
- Read-side copy for cross-domain views — When a screen needs data from several domains on every load.
Strategies
- Strangle one domain at a time When: When migrating an existing monolithic database. How: Pick the domain with the fewest cross-domain joins, copy its tables to a new database, switch reads, then switch writes, then drop the old tables. Example: Move the notifications tables out first because nothing else joins against them.
- Replace foreign keys with IDs plus checks When: Once two tables live in different databases. How: Keep the ID column, validate existence through the owning service at write time, and handle deletions with events or soft deletes. Example: Before creating an order, the Orders service asks the Catalogue service whether the product is active.
What you lose when you federate
Inside one database you get joins, foreign keys and multi-table transactions for free. Federation gives all three up at the boundary. A query like 'top customers by order value in each product category' now touches three databases and must be assembled in code, or answered from a reporting copy.
Transactions are the bigger loss. Creating an order and reserving stock used to be one ACID transaction; across two databases it becomes a distributed workflow (see Sagas in the Consistency topic). That is why the boundaries must follow real domain seams: if two tables are constantly written together, they belong in the same database.
Tradeoffs
| Decision | Upside | Downside |
|---|---|---|
| Smaller, independent databases vs cross-domain queries | Less contention, smaller working sets, independent scaling and failure isolation. | Joins, foreign keys and transactions across domains move into application code. |
| Federation vs sharding | Most queries still hit one database because each domain stays whole. | It does not help when a single domain's table outgrows one primary. |
Real World
| System | How it's used |
|---|---|
| Microservice platforms | Teams at companies like Amazon and Uber give each service its own database; this is federation enforced by organisation boundaries. |
| Monolith migrations | Companies splitting a Rails or Django monolith typically federate the database domain by domain before any sharding. |
Interview
Questions interviewers ask
- How would you split a single overloaded database?
- What breaks when users and orders live in different databases?
- When is federation not enough?
What a strong answer covers
Explain splitting by domain, the loss of joins, foreign keys and transactions at the boundary, and when a single domain still needs sharding.
Common traps
- Splitting tables that are written in the same transaction.
- Calling federation 'sharding' — it splits by domain, not by rows.
Quiz
What does federation split a database by?
- Row ranges of one table
- Business domain or function
- Read vs write traffic
- Geographic region only
Federation gives each domain (users, orders, catalogue) its own database. Splitting rows of one table is sharding.
Which capability is lost across federated databases?
- Indexes
- Replication
- Multi-database ACID transactions and joins
- Backups
Each database still has indexes, replicas and backups, but the engine can no longer join or transact across the boundary.
A single 'events' table outgrows one primary. Does federation solve it?
- Yes, always
- No — that domain needs sharding
- Yes, if you add a cache
- Only with foreign keys
Federation moves whole domains apart; it does not split a single huge table. That requires sharding.
Which tables should stay in the same database?
- Tables owned by different teams
- Tables that are written together in one transaction
- Tables with different access patterns
- Tables with different sizes
Tables written atomically together must share a database, or you turn a simple transaction into a distributed workflow.
After federation, how is 'order references a valid product' enforced?
- Foreign key across databases
- Application-level checks or events
- The load balancer
- It cannot be enforced at all
Cross-database foreign keys do not exist, so the owning services validate references and react to deletions.
Denormalization
Store redundant copies of data so hot reads avoid expensive joins, and keep those copies correct when the source changes.
Intuition
Normalized schemas store each fact once, which keeps writes simple and data consistent. But a feed, product page or dashboard may need five joins on every request. At high read volume — or after federation, when joins cross databases — those joins become the bottleneck.
Most large systems are read-heavy, often 10:1 or 100:1. Denormalization trades extra work on the rare write for cheaper reads on the frequent path. It is how feeds, counters and search indexes stay fast.
Mental Model
Decide what the read path needs and precompute it. Copy the author's name onto each post, keep a like_count column instead of counting rows, or maintain a materialized view. Every copy is a promise: when the source changes, something must update the copies. Think of it like: A restaurant menu prints each dish's price instead of telling you to look up every ingredient's cost. Reading the menu is instant, but when prices change, every menu must be reprinted.
Building Blocks
- Redundant columns: Copy frequently read fields onto the row that is read, such as author_name on posts.
- Precomputed aggregates: Store counts and totals (followers_count, cart_total) instead of computing them per request.
- Materialized views: Database-maintained or job-maintained query results that can be read like a table.
- Update propagation: Triggers, application code, or events that keep copies in sync with the source of truth.
Definitions
- Normalization
-
Organising a schema so each fact is stored exactly once.
- Avoids update anomalies.
- Reads often need joins to reassemble data.
- Denormalization
-
Deliberately duplicating data to speed up reads.
- Makes writes do more work.
- Copies can drift if updates are missed.
- Materialized view
-
The stored result of a query that is refreshed instead of recomputed on every read.
- PostgreSQL REFRESH MATERIALIZED VIEW recomputes it in full.
- Incremental refresh needs extra tooling or application logic.
- Write amplification
-
One logical change causing many physical writes.
- Renaming a user may update thousands of denormalized rows.
Patterns
- Counter columns — When counts are read far more often than they change.
- Snapshot data that should not change — When a record must reflect values at a point in time.
- Read model built from events — When the read shape differs a lot from the write shape (CQRS).
Strategies
- Same-transaction updates When: When the copy lives in the same database as the source. How: Update the source row and its copies in one transaction, so they can never disagree. Example: Insert into likes and increment posts.like_count together.
- Asynchronous propagation When: When copies live elsewhere or there are too many to update synchronously. How: Publish a change event; consumers update their copies. Reads may be briefly stale, and consumers must be idempotent. Example: A user rename event updates author_name on that user's posts in batches.
Only denormalize what you can keep correct
The cost of denormalization is not storage; it is the bug where one copy is updated and another is not. Before adding a copy, write down who updates it, when, and what happens if that update fails halfway.
Good candidates change rarely, or are allowed to be slightly stale, or are snapshots that must never change. Poor candidates change constantly and must be exact everywhere — such as an account balance copied into several tables. When in doubt, start with a cache or a materialized view you can rebuild from the source, rather than copies you must keep in sync by hand.
Tradeoffs
| Decision | Upside | Downside |
|---|---|---|
| Fast reads vs harder writes | Hot pages read one row instead of joining several tables. | Every write must update every copy, and a missed update leaves inconsistent data. |
| Synchronous vs asynchronous copy updates | Synchronous updates keep copies exact. | They make writes slower and can lock many rows; asynchronous updates are faster but briefly stale. |
Real World
| System | How it's used |
|---|---|
| Social feeds | Twitter-style timelines store denormalized post data per follower so the home feed is a single read. |
| E-commerce orders | Orders copy price and title at checkout so invoices stay correct even if the product changes later. |
Interview
Questions interviewers ask
- When would you denormalize a schema?
- How do you keep denormalized copies consistent?
- Counter column or COUNT(*) — which and why?
What a strong answer covers
Identify the hot read path, show the copy you would add, and explain exactly how and when it is updated, including failure handling.
Common traps
- Denormalizing data that changes constantly and must be exact.
- Not saying who updates the copies when the source changes.
Quiz
What does denormalization optimise for?
- Fewer writes
- Cheaper reads on hot paths
- Smaller storage
- Stronger consistency
It duplicates data so frequent reads avoid joins or aggregation, at the cost of more work on writes.
Why does an order line store unit_price instead of joining to products?
- Joins are not allowed
- To keep the price as it was at purchase time
- To save space
- To speed up writes
This is a snapshot: the order must reflect the price paid, even if the catalogue price changes later.
A user renames themselves and 50,000 posts store author_name. What is this called?
- Write amplification
- Read repair
- Replication lag
- Index bloat
One logical change triggers many physical writes because the data was duplicated.
Safest way to maintain like_count in the same database as likes?
- A nightly job
- Update both in one transaction
- Recompute it on every read
- Let the cache handle it
Inserting the like and incrementing the counter in one transaction guarantees they never disagree.
Which data is the worst candidate for denormalization?
- A product title on past orders
- A follower count shown on profiles
- An account balance that must be exact everywhere
- An author name on blog posts
Data that changes often and must be exact in every copy is the hardest to keep correct when duplicated.
SQL Tuning at Scale
Find the queries that actually cost the most, read their execution plans, and fix them before adding hardware or splitting data.
Intuition
A database at high CPU is usually not short of hardware; it is doing unnecessary work. A missing index, an N+1 query loop, a deep OFFSET page or SELECT * on wide rows can each multiply load by orders of magnitude.
Tuning is the cheapest scaling technique: no new infrastructure, no data migration, and often a 10–1000× improvement on the hottest query. Interviewers expect you to check this before proposing sharding.
Mental Model
Rank queries by total time (calls × average time), not by the slowest single query. Take the top few, run EXPLAIN ANALYZE, find where rows are scanned or sorted unnecessarily, fix that one thing, and measure again. Repeat until the database is no longer the bottleneck. Think of it like: A doctor ranks problems by impact, examines one at a time and checks the result after each treatment, instead of prescribing everything at once.
Building Blocks
- Query statistics: pg_stat_statements (PostgreSQL) or the MySQL slow log rank queries by total and average time.
- Execution plans: EXPLAIN ANALYZE shows whether the database scans the whole table, uses an index, sorts in memory or spills to disk.
- Composite and covering indexes: Indexes ordered to match WHERE and ORDER BY clauses; covering indexes include every column the query reads.
- Query shape fixes: Batching N+1 loops, keyset pagination, selecting only needed columns and avoiding functions on indexed columns.
Definitions
- N+1 query
-
Loading a list with one query, then running one extra query per item.
- 100 posts become 101 round trips.
- Fix with a join or a single WHERE id IN (...) query.
- Covering index
-
An index that contains every column a query needs.
- The database can answer from the index without reading table rows (an index-only scan).
- Keyset pagination
-
Paging with WHERE id > last_seen_id instead of OFFSET.
- OFFSET 100000 still reads and discards 100,000 rows.
- Keyset pagination stays fast on deep pages.
- Leftmost-prefix rule
-
A composite index on (a, b, c) helps queries filtering on a, or a and b, but not on b alone.
- Order index columns by how queries filter and sort.
Patterns
- Index for the access pattern — When the plan shows a sequential scan on a large table for a selective filter.
- Batch the N+1 — When an ORM issues one query per row in a loop.
- Keyset pagination — For infinite scroll or deep pages.
Strategies
- Fix by total time When: Always, when choosing what to tune first. How: Sort queries by calls × mean time. A 5 ms query run 2 million times a day costs more than a 10 s report run once. Example: pg_stat_statements shows a profile lookup consuming 40% of total database time.
- One change, then measure When: Every time you tune. How: Apply a single index or rewrite, compare the plan and latency, and keep it only if it helps. Indexes also slow writes, so remove unused ones. Example: A new index drops the query from 800 ms to 3 ms; an unused index on the same table is dropped.
Why indexes are not free
Every index must be updated on every INSERT, UPDATE and DELETE that touches its columns, and it competes for memory with the data itself. A write-heavy table with ten indexes can spend more time maintaining indexes than writing rows.
So tuning is a balance: add the few indexes that the hottest reads need, prefer one composite index that serves several queries over many single-column ones, and periodically drop indexes the statistics show are never used. Also watch for queries that cannot use an index at all, such as WHERE LOWER(email) = ? without an expression index, or a leading wildcard LIKE '%term'.
Tradeoffs
| Decision | Upside | Downside |
|---|---|---|
| More indexes vs write speed | Reads become fast lookups instead of scans. | Every write maintains every index; unused indexes are pure cost. |
| Tuning vs re-architecting | Tuning is quick, cheap and reversible. | It cannot remove a hard limit such as a single primary's write throughput. |
Real World
| System | How it's used |
|---|---|
| PostgreSQL in production | Teams enable pg_stat_statements and auto_explain to find the few queries that dominate database time. |
| Rails and Django apps | Tools such as Bullet detect N+1 queries in development before they reach production. |
Interview
Questions interviewers ask
- The database CPU is at 90%. Walk me through your first steps.
- Why is OFFSET pagination slow on deep pages?
- When does adding an index make things worse?
What a strong answer covers
Describe measuring by total time, reading plans, specific fixes (composite indexes, batching, keyset pagination) and the write cost of indexes.
Common traps
- Proposing sharding before checking for missing indexes or N+1 queries.
- Adding an index on every column 'just in case'.
Quiz
Which query should you tune first?
- The slowest single query
- The one with the highest total time (calls × average)
- The newest query
- The one with the most joins
A fast query that runs millions of times can dominate database load more than a rare slow one.
An index on (user_id, created_at) helps which query?
- WHERE created_at > ? only
- WHERE user_id = ? ORDER BY created_at
- WHERE email = ?
- WHERE LOWER(name) = ?
Composite indexes follow the leftmost-prefix rule: filtering on user_id and sorting by created_at matches the index order.
Why is OFFSET 100000 LIMIT 20 slow?
- It locks the table
- The database still reads and discards 100,000 rows
- It disables indexes
- It runs on a replica
OFFSET skips rows by reading them. Keyset pagination jumps straight to the cursor position using an index.
A page loads 50 posts and runs 51 queries. What is the fix?
- Add a replica
- Batch the per-post queries into one
- Increase the connection pool
- Add more app servers
That is an N+1 pattern; one query with a join or WHERE id IN (...) replaces the 50 extra round trips.
What is the main cost of an index?
- Slower reads
- Extra work on every write and extra memory
- It disables transactions
- It requires sharding
Writes must update every index, and indexes compete with data for memory.
Resharding & Cross-Shard Queries
Handle what basic sharding leaves out: queries that span shards, uneven growth, and moving data to new shards without downtime.
Intuition
Sharding (Phase 3) spreads rows across databases by a shard key. Two problems arrive later. Some queries need data from every shard, and shards grow unevenly until one is full or overloaded and must be split — while the system keeps serving traffic.
A sharding design is only as good as its answer to 'what happens in two years?'. Interviewers probe resharding and cross-shard queries because they decide whether the design survives growth.
Mental Model
Pick the shard key so that most queries hit exactly one shard. Plan for growth up front: map data to many logical shards and place several logical shards on each physical server. Resharding then becomes moving whole logical shards between servers instead of re-splitting rows. Think of it like: Pack books into many small labelled boxes instead of a few huge ones. When a shelf fills up, you move a few boxes to a new shelf; you never unpack and re-sort every book.
Building Blocks
- Shard key choice: A high-cardinality key that most queries include, such as tenant_id or user_id, so queries stay on one shard.
- Logical shards: Many more shards than servers (for example 4,096 logical on 16 physical), so data can be rebalanced by moving shards.
- Shard map: A lookup from logical shard to physical server, stored in a config service and cached by routers.
- Online migration: Copy, catch up from the change log, switch routing, then delete — without stopping writes.
Definitions
- Scatter-gather query
-
A query sent to every shard whose results are merged by the caller.
- Latency is set by the slowest shard.
- Cost grows with the number of shards.
- Resharding
-
Changing how data is distributed across shards.
- Needed when shards grow unevenly or the cluster grows.
- Must avoid downtime and lost writes.
- Logical shard
-
A fixed partition of the key space that can be moved as a unit between physical servers.
- Also called a virtual shard or bucket.
- Global secondary index
-
An index partitioned by a different key than the data, so lookups by that key hit one index shard.
- Keeping it updated across shards is asynchronous in most systems.
Patterns
- Shard by the dominant access key — When one entity scopes most queries.
- Lookup table for alternate keys — When some queries use a key other than the shard key.
- Separate analytics store — When reports need data from every shard.
Strategies
- Copy, catch up, cut over When: Moving a shard to a new server without downtime. How: Snapshot the shard to the new server, replay changes from the replication log until it is caught up, briefly pause writes for that shard, switch the shard map, resume writes and delete the old copy. Example: Vitess and similar tools automate this workflow for MySQL.
- Split hot shards, don't rehash everything When: When one shard grows much faster than the others. How: Divide only the hot shard's key range in two and move one half, leaving all other shards untouched. Example: A range-sharded table splits the shard holding the newest IDs.
Why hash(key) % N makes resharding painful
If the shard is chosen as hash(key) % N, changing N from 8 to 10 moves most keys to a different shard, so nearly all data must be copied at once. That is why mature systems avoid tying the formula to the number of servers.
The two common answers are a fixed, large number of logical shards mapped to servers through a table, or consistent hashing (next topic), where adding a node moves only about 1/N of the keys. Both turn resharding into moving small, well-defined pieces of data.
Tradeoffs
| Decision | Upside | Downside |
|---|---|---|
| Many logical shards vs simplicity | Rebalancing becomes moving whole shards; no row-level re-splitting. | More metadata, more connections and more operational tooling. |
| Scatter-gather vs a separate read store | Scatter-gather needs no extra system. | It gets slower and costlier with every shard; a warehouse or index scales better for cross-shard queries. |
Real World
| System | How it's used |
|---|---|
| Vitess (YouTube, Slack) | Shards MySQL by keyspace ranges and performs online resharding with copy, catch-up and cutover. |
| Mapped thousands of logical shards onto far fewer PostgreSQL servers so they could be moved as load grew. |
Interview
Questions interviewers ask
- How would you choose a shard key for this system?
- How do you answer a query that needs data from every shard?
- How do you add shards without downtime?
What a strong answer covers
Justify the shard key from access patterns, handle alternate-key lookups, describe logical shards plus online migration, and keep reporting off the production shards.
Common traps
- Using hash(key) % number_of_servers with no plan for growth.
- Ignoring queries that do not include the shard key.
Quiz
What makes a good shard key?
- A low-cardinality status field
- A key most queries include, with many distinct values
- The row's creation timestamp only
- A random value unrelated to queries
Queries that include the shard key hit one shard, and high cardinality spreads data evenly.
Why map many logical shards onto fewer servers?
- To avoid indexes
- So rebalancing moves whole shards instead of re-splitting rows
- To remove replication
- To allow cross-shard joins
Moving a logical shard between servers is a contained copy; the key-to-shard mapping never changes.
What sets the latency of a scatter-gather query?
- The fastest shard
- The average shard
- The slowest shard
- The router's CPU
The caller must wait for every shard, so the slowest response determines the total.
Using hash(key) % N, what happens when N goes from 8 to 10?
- Only new keys move
- About 20% of keys move
- Most keys map to a different shard
- Nothing changes
Changing the modulus changes the result for most keys, forcing a near-total data move.
Correct order for moving a shard online?
- Switch routing, then copy data
- Copy snapshot, replay changes, cut over, delete old copy
- Delete the old copy, then copy
- Stop all writes for the whole cluster
Snapshot and catch-up happen while traffic continues; only the brief cutover pauses writes for that shard.
Practice database scaling techniques in PRISM
Concepts stick when you watch them fail. Build an architecture that depends on database scaling techniques, push traffic through it in the PRISM simulator, and see the latency and error rates change as you adjust the design.