- A relational (SQL) database stores data in tables with a fixed schema, enforces relationships, and offers ACID transactions — ideal for data where correctness matters (accounts, usernames, the follow relationship).
- Isolation levels (read uncommitted → read committed → repeatable read → serializable) trade speed for protection against anomalies: dirty reads, non-repeatable reads, phantoms and lost updates.
- NoSQL is a family of six main shapes — key-value, document, wide-column, graph, time-series and search — each optimised for a specific access pattern, usually by giving up joins and strict consistency (BASE) in exchange for horizontal scale.
- Wide-column stores like Cassandra are modelled query-first: you choose a partition key (which node) and clustering key (sort order inside the partition) for each query, and you denormalise freely.
- Real systems use polyglot persistence. Instagram, per public engineering talks, runs sharded PostgreSQL for its relational core (users, relationships) and Cassandra for high-write data such as feeds and activity, plus Redis, Elasticsearch and object storage.
- NewSQL (Spanner, CockroachDB) tries to give you SQL + ACID and horizontal scale, at the cost of latency and operational complexity.
1. The problem: why can't Instagram just use "a database"?
Think about the different kinds of data a single Instagram session touches:
- Your account — username, email, password hash, privacy setting. Small, read often, must never be wrong: two people can't both own
@alice. - Follow relationships — billions of edges between users. You need "who does Alice follow?" and "who follows Alice?" quickly.
- Posts, likes and comments — hundreds of millions of new rows per day. Likes arrive in floods when something goes viral.
- Your feed and activity inbox — a constantly appended list per user, written far more often than it's read in full.
- Search — "find accounts whose name sounds like selena", which no ordinary table lookup can do.
- Photos and videos — petabytes of large immutable blobs that don't belong in a database at all (they go to object storage).
Each of these has a different shape (rows, edges, append-only logs, text), a different read/write ratio, different consistency needs, and a different scale. No single database engine is best at all of them. Picking a database is therefore not a matter of taste; it's a matter of matching the engine's strengths to each workload's access pattern. That is what this lesson teaches you to do.
2. The relational model
2.1 Tables, rows, keys and joins
The relational model (Edgar Codd, 1970) organises data into tables (formally relations). Each table has a fixed set of named, typed columns; each row is one record. Key vocabulary:
- Primary key (PK) — a column (or set of columns) that uniquely identifies a row, e.g.
users.id. The database guarantees no two rows share it. - Foreign key (FK) — a column that references another table's primary key, e.g.
posts.user_id → users.id. The database can refuse a post whoseuser_iddoesn't exist (this is called referential integrity). - Constraint — a rule the database enforces on every write:
NOT NULL,UNIQUE,CHECK (like_count >= 0). - Schema — the declared structure of all tables. Relational databases are schema-on-write: data must match the schema when it is written. (Most NoSQL stores are closer to schema-on-read: the application interprets whatever is stored.)
- Join — combining rows from several tables at query time, e.g. "posts together with their author's username".
- SQL — the declarative query language: you describe what you want; the query planner decides how to get it (which indexes to use, which join algorithm).
2.2 Why the relational model has lasted 50+ years
- Flexible querying. Because data is normalised and SQL is declarative, you can answer questions nobody anticipated when the schema was designed ("how many users who joined in 2019 commented on a post in Brazil?").
- Strong guarantees. Constraints and transactions keep data correct even with concurrent writers and crashes.
- Mature tooling. Backups, replication, migrations, monitoring, ORMs (Instagram began life as a Django app talking to PostgreSQL through Django's ORM).
The main weakness is that a classic relational database is designed to run on one machine (plus replicas). Joins, foreign keys and multi-row transactions are cheap when all rows are on the same disk and expensive when they live on different servers. Scaling writes beyond one machine requires sharding, and sharding breaks many of those conveniences.
3. ACID, explained with concrete examples
A transaction is a group of reads and writes that the database treats as one logical unit. ACID is the set of guarantees a transactional database makes about it.
3.1 Atomicity — all or nothing
Example: when Bob likes a post, we must (1) insert a row into likes and (2) increment posts.like_count. If the server crashes between the two steps, without atomicity we'd have a like that isn't counted (or a count with no like).
BEGIN;
INSERT INTO likes (post_id, user_id, created_at) VALUES (9001, 42, now());
UPDATE posts SET like_count = like_count + 1 WHERE id = 9001;
COMMIT; -- both changes become visible together, or (on error/crash) neither does
If anything fails before COMMIT — a constraint violation, a crash, an explicit ROLLBACK — the database undoes every change in the transaction. "Atomic" here means indivisible, not "concurrent-safe" (that's isolation).
3.2 Consistency — the rules always hold
Consistency means a transaction takes the database from one valid state to another, where "valid" is defined by your invariants: constraints, foreign keys, uniqueness. Examples:
UNIQUE (username)— two concurrent sign-ups for@alice: exactly one succeeds; the other gets a unique-violation error.PRIMARY KEY (post_id, user_id)onlikes— a double-tap that sends the request twice cannot create two likes.FOREIGN KEY (post_id) REFERENCES posts(id)— you can't like a post that was deleted a millisecond earlier.
3.3 Isolation — concurrent transactions don't trip over each other
Thousands of transactions run at once. Isolation means each behaves as if it were running alone. Perfect isolation (serializability) is expensive, so databases offer weaker isolation levels — the next section covers them in detail.
3.4 Durability — committed means committed
Once COMMIT returns success, the data survives a crash or power loss. Databases achieve this with a write-ahead log (WAL): before modifying data pages, the change is appended to a sequential log file and fsync-ed to disk. After a crash, the database replays the WAL to recover. In distributed systems, durability also means "written to more than one machine" — see Replication.
Client Database engine Disk | COMMIT | | |--------------------------> | 1. append change records to WAL | | |------------------------------------>| wal/000123 | | 2. fsync(WAL) (data is now safe) | | |<------------------------------------| | <------ "COMMIT OK" ------ | | | | 3. later: flush dirty data pages | | | in the background (checkpoint) | | |------------------------------------>| table files Crash after step 2? → on restart, replay WAL from last checkpoint → nothing lost.
4. Transactions and isolation levels
To understand isolation levels you first need the anomalies — the weird things that can happen when transactions interleave.
4.1 The four classic anomalies
🫥 Dirty read
T2 reads data that T1 has written but not yet committed. If T1 rolls back, T2 acted on data that never existed.
🔁 Non-repeatable read
T1 reads a row twice and gets different values, because T2 committed an update in between.
👻 Phantom read
T1 runs the same range query twice and gets a different set of rows, because T2 inserted or deleted matching rows.
💥 Lost update
T1 and T2 both read a value, both compute a new value from it, both write. One write silently overwrites the other.
Dirty read in Instagram terms — a moderation job reads a post that is being created and fails:
T1 (Upload Service) T2 (Moderation job)
BEGIN
INSERT post 777 (caption "hi")
SELECT * FROM posts WHERE id = 777
→ sees post 777 ← DIRTY: not committed!
sends "new post" notification
ROLLBACK (media upload failed)
→ post 777 never existed, but a notification about it was already sent.
Lost update — two like requests processed by an application that does read-modify-write in code:
T1 (like from Bob) T2 (like from Carol)
SELECT like_count FROM posts WHERE id=9 → 100
SELECT like_count FROM posts WHERE id=9 → 100
UPDATE posts SET like_count = 101
COMMIT
UPDATE posts SET like_count = 101
COMMIT
Final value: 101. Expected: 102. Bob's like was LOST.
Non-repeatable read — a report reads Alice's follower count at the start and end of a transaction and gets 1,000 then 1,001, so its totals don't add up. Phantom — a job counts "comments on post 9 in the last hour" twice in one transaction and gets 50 then 51 because a new comment row appeared.
There's also a subtler anomaly, write skew: two transactions each read an overlapping set of rows, make a decision based on it, and write to different rows, together breaking an invariant. Classic example: "at least one admin must remain on a group" — two admins each check "there are 2 admins, so I can leave" and both leave. Only serializable isolation prevents write skew in general.
4.2 The four standard isolation levels
| Isolation level | Dirty read | Non-repeatable read | Phantom | Lost update | Typical implementation |
|---|---|---|---|---|---|
| Read uncommitted | ❌ possible | ❌ possible | ❌ possible | ❌ possible | No read protection. Rarely used (PostgreSQL silently treats it as read committed). |
| Read committed | ✅ prevented | ❌ possible | ❌ possible | ❌ possible | Each statement sees a fresh snapshot of committed data. Default in PostgreSQL, Oracle, SQL Server. |
| Repeatable read | ✅ prevented | ✅ prevented | ⚠️ allowed by the SQL standard; prevented by snapshot-based engines (PostgreSQL) | ⚠️ PostgreSQL detects and aborts; MySQL InnoDB does not for plain read-then-write | Whole transaction sees one snapshot (snapshot isolation). Default in MySQL InnoDB. |
| Serializable | ✅ prevented | ✅ prevented | ✅ prevented | ✅ prevented (and write skew) | Result equals some serial order. Via strict 2-phase locking, actual serial execution, or Serializable Snapshot Isolation (PostgreSQL SSI, CockroachDB). |
Read the table top to bottom as "each level adds protection and adds cost". Stronger levels either block more (locks) or abort more (optimistic engines raise a serialization failure that the application must retry).
4.3 How databases implement isolation: locks and MVCC
- Locking. A transaction takes shared locks for reads and exclusive locks for writes. Two-phase locking (2PL) — acquire locks as you go, release them only at commit — gives serializability but readers block writers and vice versa, and deadlocks become possible (the database detects cycles and aborts one victim).
- MVCC (multi-version concurrency control). Instead of overwriting a row, the database keeps multiple versions tagged with the transaction ID that created them. A reader sees the versions that were committed as of its snapshot. Readers never block writers and writers never block readers. PostgreSQL, MySQL InnoDB, Oracle and CockroachDB all use MVCC. Old versions are later cleaned up (PostgreSQL's
VACUUM).
4.4 Fixing lost updates in practice
Most production apps run at read committed for performance, so you must know the targeted fixes for lost updates:
- Atomic update in the database (best):
UPDATE posts SET like_count = like_count + 1 WHERE id = 9;— the read and the write happen inside one statement under a row lock. - Explicit row lock:
SELECT like_count FROM posts WHERE id = 9 FOR UPDATE;— the second transaction waits until the first commits. - Optimistic concurrency control (compare-and-set): add a
versioncolumn;UPDATE posts SET caption = 'new', version = 8 WHERE id = 9 AND version = 7;— if 0 rows are affected, someone else won; re-read and retry. Great for rarely-conflicting edits like editing a caption. - Raise the isolation level to repeatable read (PostgreSQL) or serializable and retry on serialization failure.
5. Normalization vs denormalization
5.1 Normalization: store each fact once
Normalization organises tables so that every fact is stored in exactly one place, eliminating update anomalies (changing a fact in one place but forgetting another copy). The first three normal forms, informally:
- 1NF — every column holds a single atomic value (no comma-separated list of tags in one cell; use a separate
post_tagstable). - 2NF — every non-key column depends on the whole primary key (in
likes(post_id, user_id, …)don't storepost_caption, which depends only onpost_id). - 3NF — non-key columns depend only on the key, not on other non-key columns (don't store
author_usernameinposts; it depends onuser_id, which lives inusers).
The schema in the diagram above is normalised: a username lives only in users; to show it next to a post you join.
5.2 Denormalization: store facts where you read them
Denormalization deliberately duplicates or precomputes data to make reads cheap. Instagram-style examples:
posts.like_count— a precomputed count instead ofSELECT COUNT(*) FROM likes WHERE post_id = ?over millions of rows on every feed render.- Storing
author_usernameandauthor_avatar_urlinside each feed entry so the feed can be rendered without a join to a different shard. - Storing the same comment in
comments_by_postandcomments_by_usertables in Cassandra, one per query.
| Normalized | Denormalized | |
|---|---|---|
| Reads | Need joins — slower, and hard across shards | Single lookup — fast and shard-local |
| Writes | One place to update — cheap and simple | Must update every copy (fan-out writes) |
| Consistency | Automatically consistent | Copies can drift; needs background repair or events |
| Storage | Minimal | More (duplicates) |
| Flexibility | Any new query possible | Optimised for known queries only |
| Best for | Write-heavy or correctness-critical data, ad-hoc analytics | Read-heavy, latency-critical paths (feeds, profiles) |
Rule of thumb: start normalised, denormalise specific hot read paths, and keep the denormalised copies updated asynchronously through events (e.g. a username_changed event on Kafka). Accept that a renamed user's old name may show in a few feed entries for a short while.
6. The NoSQL families
"NoSQL" ("not only SQL") isn't one technology; it's a label for databases that drop parts of the relational model — usually joins, multi-row transactions and fixed schemas — to gain horizontal scalability, high write throughput, flexible schemas or a more natural data model for a specific problem. There are six families you should know.
🔑 Key-value
A giant distributed hash map: get(key), put(key, value). The value is opaque. Fastest and simplest. Redis, Memcached, DynamoDB, RocksDB, etcd.
📄 Document
Values are JSON-like documents with nested fields, queryable by field and indexable. Schema-flexible. MongoDB, Couchbase, Firestore, DynamoDB (also).
🧱 Wide-column
Rows grouped into partitions by a partition key, sorted by clustering columns; built on LSM trees for huge write throughput. Cassandra, ScyllaDB, HBase, Bigtable.
🕸️ Graph
Nodes and edges are first-class; queries traverse relationships ("friends of friends who like X"). Neo4j, Amazon Neptune, JanusGraph; Facebook's TAO is a graph-shaped cache over MySQL.
📈 Time-series
Append-mostly (series, timestamp, value) points, heavy compression, downsampling and retention. Prometheus, InfluxDB, TimescaleDB, Facebook's Gorilla.
🔎 Search engine
Inverted indexes over text, relevance scoring, fuzzy matching, facets. Usually a secondary copy of data. Elasticsearch, OpenSearch, Solr.
| Family | Data model | Good at | Bad at | Instagram-style use case |
|---|---|---|---|---|
| Key-value | key → blob | O(1) lookups, caching, sessions, counters, extreme throughput | Queries by anything but the key; relationships | Session tokens, feed cache & counters in Redis, rate-limit buckets |
| Document | key → JSON document | Self-contained aggregates read together; evolving schemas | Many-to-many relationships, cross-document transactions (limited) | User settings blob, a post with its media list and tags as one document |
| Wide-column | (partition key, clustering key) → columns | Massive write rates, time-ordered data per entity, linear scale-out, multi-DC | Ad-hoc queries, joins, aggregations, strong consistency by default | Feed timelines, activity/notification inbox, comments by post, messages |
| Graph | nodes + typed edges | Multi-hop traversals, recommendations, fraud rings | Huge write rates of simple edges, sharding (traversals cross shards) | "Suggested for you" (friends-of-friends); the social graph |
| Time-series | (metric, tags, ts) → value | Ingesting millions of points/sec, range aggregations, compression | Updates, arbitrary entity lookups | Service metrics and dashboards in observability |
| Search | documents + inverted index | Full-text, prefix/fuzzy, relevance ranking, facets | Being the source of truth, transactions, fast exact updates | User/hashtag/place search |
JSONB document type, full-text search and (via extensions) time-series and geospatial support; Redis has sorted sets, streams and search modules. Choose by access pattern and scale, not by label.7. BASE: the NoSQL counterpart to ACID
Many distributed NoSQL stores describe their guarantees as BASE:
- Basically Available — the system answers requests even when some nodes or network links fail, possibly with stale data, rather than refusing.
- Soft state — the state of replicas may change over time even without new input, as background replication and repair converge them.
- Eventual consistency — if writes stop, all replicas eventually return the same value. There's no promise about when (in practice usually milliseconds).
BASE is the natural choice when you prefer availability during network partitions (the "AP" side of CAP). Example: if a like takes 500 ms to appear for a user on another continent, nobody is harmed; if Instagram refused all likes during a network blip, millions would be annoyed.
| ACID | BASE | |
|---|---|---|
| Priority | Correctness | Availability & scale |
| Consistency | Strong — reads see committed latest data | Eventual — reads may be stale for a while |
| Under a partition | May reject requests | Keeps serving; reconciles later |
| Conflict handling | Prevented by locks/serialization | Resolved after the fact (last-write-wins, CRDTs) |
| Example | Username registration, payments | Like counts, feed entries, view counts |
It isn't binary: Cassandra lets you choose a consistency level per query (ONE, QUORUM, ALL), and with QUORUM reads and writes you get read-your-latest-write behaviour. The maths (R + W > N) is in Replication.
8. Wide-column deep dive: Cassandra
Apache Cassandra (originally built at Facebook for inbox search, combining Amazon Dynamo's distribution with Google Bigtable's data model) is the wide-column store you are most likely to be asked about — and one Instagram has publicly described using heavily.
8.1 The data model: partition key + clustering key
A Cassandra table's PRIMARY KEY has two parts:
- Partition key (the first part, in inner parentheses) — hashed to decide which nodes store the data (via consistent hashing on a token ring). All rows sharing a partition key form one partition, stored together on the same replicas.
- Clustering key (the remaining columns) — decides the sort order of rows inside a partition. Rows are physically stored sorted by it, so range scans within a partition ("the 20 newest") are sequential and fast.
CREATE TABLE comments_by_post (
post_id bigint,
comment_id bigint, -- time-sortable ID, newer = larger
user_id bigint,
text text,
PRIMARY KEY ((post_id), comment_id) -- partition key: post_id; clustering key: comment_id
) WITH CLUSTERING ORDER BY (comment_id DESC);
-- The query this table exists to serve: one partition, sequential read, newest first
SELECT comment_id, user_id, text FROM comments_by_post
WHERE post_id = 9001 LIMIT 20;
Token ring (hash of partition key → node) Inside node B, partition post_id=9001
(rows sorted by clustering key DESC)
hash(9001) = 0x7A.. ──► Node B (+ 2 replicas) +-----------------+---------+-------------+
hash(9002) = 0x13.. ──► Node A | comment_id | user_id | text |
hash(9003) = 0xC4.. ──► Node D +-----------------+---------+-------------+
| 7012..998 (new) | 42 | "wow!" |
| 7012..871 | 77 | "where is.."|
| 7011..003 | 15 | "first!" |
| ... could be millions of rows ... |
+-----------------+---------+-------------+
8.2 Query-first modelling
In SQL you model entities and figure out queries later. In Cassandra you do the opposite: list your queries first, then create one table per query, shaped so each query hits exactly one partition. The rules:
- Every query must specify the full partition key with equality. Cassandra won't efficiently answer "comments by user 42" from
comments_by_post— that would need to scan every partition on every node. - Range filters and ordering only on clustering columns, in their declared order.
- No joins, so denormalise. Need comments by user too? Create
comments_by_userwithPRIMARY KEY ((user_id), comment_id)and write each comment to both tables (a logged batch ensures both writes eventually apply). - Writes are cheap, reads must be planned. Cassandra's LSM-tree storage (see Indexing & Storage Engines) appends writes sequentially, so duplicating data is the accepted price for single-partition reads.
8.3 Wide partitions and bucketing
A partition lives entirely on its replica nodes, so it can't grow forever. Guidance commonly cited for Cassandra is to keep partitions under roughly 100 MB and a few hundred thousand rows. A celebrity's activity feed or a viral post's comments can blow past that, causing slow reads, compaction pain, GC pauses and a hot partition overloading three nodes while the rest idle.
The fix is bucketing: add a component to the partition key that splits one logical entity into many physical partitions.
CREATE TABLE activity_by_user (
user_id bigint,
day_bucket int, -- e.g. 20261023 (one partition per user per day)
activity_id timeuuid,
actor_id bigint,
type text, -- 'like' | 'comment' | 'follow' | 'mention'
post_id bigint,
PRIMARY KEY ((user_id, day_bucket), activity_id)
) WITH CLUSTERING ORDER BY (activity_id DESC)
AND default_time_to_live = 7776000; -- 90 days, old activity expires automatically
Reading "latest 50 activities" now means: query today's bucket; if fewer than 50, query yesterday's, and so on. Worked example: if a celebrity receives 2 million activity events a day at ~200 bytes each, a daily bucket is ~400 MB — still too big, so you would bucket by hour (≈17 MB) or add a hash sub-bucket (user_id, day, shard 0..15). Bucket size is a deliberate design decision based on your write rate.
8.4 Other Cassandra facts worth knowing
- Writes are upserts.
INSERTon an existing key overwrites it; there's no "row already exists" error. Conflicts are resolved by last-write-wins using per-cell timestamps. - Deletes write tombstones. A delete is a marker that hides older data until compaction removes it after
gc_grace_seconds. Queue-like "insert then delete" patterns create tombstone storms that make reads crawl. - Lightweight transactions (
INSERT … IF NOT EXISTS) use Paxos for compare-and-set on a single partition — about 4 round trips, so use sparingly (e.g. claiming a unique handle). - Counters are a special column type for distributed increments, but they're not idempotent (a retried increment may double count).
- Secondary indexes and materialized views exist but are generally discouraged for high-cardinality data at scale; prefer explicit denormalised tables.
- Multi-datacenter replication is built in:
NetworkTopologyStrategywith e.g. RF=3 per region.
9. SQL vs NoSQL: how to decide
| Question about the workload | Points to SQL (PostgreSQL/MySQL) | Points to NoSQL |
|---|---|---|
| Do you need multi-row / multi-table transactions? | Yes — money, inventory, unique usernames | No, or only within one key/partition |
| Are relationships rich and queried in many ways? | Yes — joins and ad-hoc queries | No — few, known access patterns |
| Write volume? | Fits on one primary (or a manageable number of shards) | Hundreds of thousands to millions of writes/sec |
| Data volume? | GBs to tens of TBs | Hundreds of TBs to PBs |
| Schema stability? | Well understood, stable | Evolving, sparse or heterogeneous |
| Consistency need? | Strong, immediately | Eventual is acceptable |
| Availability across regions during partitions? | Can tolerate failover pauses | Must keep accepting writes everywhere |
| Data shape? | Tabular entities | Aggregates (document), edges (graph), logs/timelines (wide-column), text (search) |
| Team skills & tooling? | Everyone knows SQL; rich ecosystem | Needs specialised modelling and ops knowledge |
10. Polyglot persistence
Polyglot persistence means using several database technologies in one system, each for the data it suits best. Each microservice owns its data store, and data is kept in sync between stores via events on the event bus or change data capture (CDC), rather than distributed transactions.
Benefits: each workload gets the right tool and scales independently; a problem in the search cluster doesn't take down sign-ups. Costs: more systems to operate, monitor, back up and secure; data is duplicated and only eventually consistent across stores; engineers must learn several modelling styles. A good principle: one store is the source of truth for each piece of data; every other copy is a derived, rebuildable view.
11. NewSQL: SQL + ACID + horizontal scale
NewSQL databases keep the relational model and serializable/strong transactions but distribute data automatically across many machines. They split tables into ranges (shards), replicate each range with a consensus protocol (Paxos or Raft), and run distributed transactions under the hood.
Google Spanner
Globally distributed, externally consistent (linearizable) transactions. Uses TrueTime — GPS and atomic clocks giving a bounded clock uncertainty — and "commit wait" to order transactions globally. Paxos per split. Available as Cloud Spanner.
CockroachDB
Open-source, Spanner-inspired, PostgreSQL wire-compatible. Data split into ~512 MB ranges, each a Raft group of 3+ replicas. Serializable by default. Uses hybrid logical clocks instead of atomic clocks.
TiDB / YugabyteDB
TiDB: MySQL-compatible, Raft-replicated TiKV storage. YugabyteDB: PostgreSQL-compatible query layer on a Raft-replicated document store.
Vitess & Citus (related)
Not NewSQL engines but sharding layers over MySQL / PostgreSQL that give similar scale-out with the familiar engine. See Sharding.
Trade-offs: every write must reach a consensus majority (often across zones or regions), so write latency is higher than a single-node database — tens of ms cross-region instead of ~1 ms locally. Cross-range transactions cost extra round trips. Operations and cost are higher than a single PostgreSQL. They shine when you need relational semantics and global strong consistency at a scale beyond one machine (financial ledgers, inventory, global user directories) and don't want to build sharding yourself. Instagram's PostgreSQL sharding predates mature NewSQL options; a team designing from scratch today might evaluate one for the user/relationship core.
12. Example schemas: SQL and CQL
12.1 Relational schema (PostgreSQL DDL)
CREATE TABLE users (
id BIGINT PRIMARY KEY, -- 64-bit time-sortable ID (see id-generation)
username VARCHAR(30) NOT NULL UNIQUE,
email VARCHAR(255) NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
full_name VARCHAR(100),
bio VARCHAR(150),
is_private BOOLEAN NOT NULL DEFAULT false,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE posts (
id BIGINT PRIMARY KEY,
user_id BIGINT NOT NULL REFERENCES users(id),
caption VARCHAR(2200),
location_id BIGINT,
like_count INTEGER NOT NULL DEFAULT 0 CHECK (like_count >= 0), -- denormalized
comment_count INTEGER NOT NULL DEFAULT 0 CHECK (comment_count >= 0),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
deleted_at TIMESTAMPTZ -- soft delete
);
CREATE INDEX posts_user_recent ON posts (user_id, created_at DESC); -- profile grid
CREATE TABLE media (
id BIGINT PRIMARY KEY,
post_id BIGINT NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
position SMALLINT NOT NULL, -- order inside a carousel
media_type VARCHAR(10) NOT NULL CHECK (media_type IN ('photo', 'video')),
storage_key TEXT NOT NULL, -- object-storage key, not the bytes
width INTEGER, height INTEGER, duration_ms INTEGER,
UNIQUE (post_id, position)
);
CREATE TABLE likes (
post_id BIGINT NOT NULL REFERENCES posts(id),
user_id BIGINT NOT NULL REFERENCES users(id),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (post_id, user_id) -- one like per user per post
);
CREATE INDEX likes_by_user ON likes (user_id, created_at DESC);
CREATE TABLE comments (
id BIGINT PRIMARY KEY,
post_id BIGINT NOT NULL REFERENCES posts(id),
user_id BIGINT NOT NULL REFERENCES users(id),
parent_comment_id BIGINT REFERENCES comments(id), -- replies
text VARCHAR(2200) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX comments_by_post ON comments (post_id, created_at DESC);
CREATE TABLE follows (
follower_id BIGINT NOT NULL REFERENCES users(id),
followee_id BIGINT NOT NULL REFERENCES users(id),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (follower_id, followee_id), -- "whom do I follow?"
CHECK (follower_id <> followee_id)
);
CREATE INDEX follows_reverse ON follows (followee_id, follower_id); -- "who follows me?"
A typical relational query — the 12 most recent posts for a profile grid, with a cover image each:
SELECT p.id, p.like_count, m.storage_key
FROM posts p
JOIN media m ON m.post_id = p.id AND m.position = 0
WHERE p.user_id = 42 AND p.deleted_at IS NULL
ORDER BY p.created_at DESC
LIMIT 12;
users and posts are sharded by user_id, a like by user A (shard 3) on a post by user B (shard 812) can't have both foreign keys enforced by one database, and the likes table must live on one shard (usually the post's). Large systems keep FKs within a shard and enforce cross-shard references in application code.12.2 Wide-column schema (Cassandra CQL), one table per query
-- Q1: a user's posts, newest first (profile grid)
CREATE TABLE posts_by_user (
user_id bigint,
post_id bigint, -- time-sortable, so DESC = newest first
caption text,
media_keys list<text>,
created_at timestamp,
PRIMARY KEY ((user_id), post_id)
) WITH CLUSTERING ORDER BY (post_id DESC);
-- Q2: fetch one post by id (e.g. from a feed entry or a shared link)
CREATE TABLE posts_by_id (
post_id bigint PRIMARY KEY,
user_id bigint,
caption text,
media_keys list<text>,
created_at timestamp
);
-- Q3: who liked a post? / did user X like post P? (single partition, clustering lookup)
CREATE TABLE likes_by_post (
post_id bigint,
user_id bigint,
liked_at timestamp,
PRIMARY KEY ((post_id), user_id)
);
-- Q4: counts shown under each post
CREATE TABLE post_counters (
post_id bigint PRIMARY KEY,
like_count counter,
comment_count counter
);
-- Q5: comments on a post, newest first
CREATE TABLE comments_by_post (
post_id bigint,
comment_id bigint,
user_id bigint,
text text,
PRIMARY KEY ((post_id), comment_id)
) WITH CLUSTERING ORDER BY (comment_id DESC);
-- Q6 / Q7: following and followers lists (the same edge written twice)
CREATE TABLE following_by_user (
user_id bigint, followee_id bigint, since timestamp,
PRIMARY KEY ((user_id), followee_id)
);
CREATE TABLE followers_by_user (
user_id bigint, follower_id bigint, since timestamp,
PRIMARY KEY ((user_id), follower_id)
);
-- Q8: a user's home-feed timeline (post IDs pushed by fan-out workers), bucketed by month
CREATE TABLE feed_by_user (
user_id bigint,
month int, -- e.g. 202610
post_id bigint,
author_id bigint,
PRIMARY KEY ((user_id, month), post_id)
) WITH CLUSTERING ORDER BY (post_id DESC)
AND default_time_to_live = 2592000; -- keep 30 days of timeline
Notice the differences from the SQL version: there are no foreign keys, no joins, the follow edge is written twice, posts are written twice, and counters live in their own table. Every table is designed around exactly one query, and every query hits exactly one partition.
13. How Instagram uses databases
- PostgreSQL from day one. Instagram launched in 2010 as a Django application on PostgreSQL. Its engineering blog post on sharding and IDs describes splitting data into thousands of logical shards, each a PostgreSQL schema, mapped onto a smaller number of physical database servers, so shards can be moved between machines without re-hashing (details in Sharding).
- 64-bit IDs generated inside PostgreSQL. The same post describes IDs made of 41 bits of milliseconds since a custom epoch, 13 bits of logical shard ID, and 10 bits of a per-shard sequence, generated by a PL/pgSQL function — so IDs are unique, roughly time-sortable and encode their shard. See Unique ID Generation.
- Read replicas for PostgreSQL to scale reads, and later multiple data centers (Replication, Multi-Region).
- Cassandra for high-write data. Public talks (e.g. at Cassandra Summit) describe Instagram moving workloads such as the feed, activity/news inbox and direct-messaging-related data to Cassandra for write scalability and lower cost, and Instagram engineers later built Rocksandra, swapping Cassandra's storage engine for RocksDB to cut tail latency.
- Redis and Memcached for caching and fast in-memory structures; Elasticsearch-style search for discovery; and media on blob storage behind the CDN. After joining Facebook, Instagram also adopted parts of Facebook's infrastructure; Facebook's own social data famously lives in TAO, a graph-shaped cache over sharded MySQL.
In this course's reference architecture, that becomes:
| Data | Store | Why |
|---|---|---|
| Users, credentials, usernames, settings | PostgreSQL (sharded by user_id + read replicas) | Uniqueness constraints and ACID transactions; moderate write rate; rich queries for support/admin tools. |
| Post / comment / story metadata | Cassandra, partitioned by user_id (post_id for comments) | Very high write volume; reads are "by user, newest first" — perfect partition/clustering fit; TTL for stories. |
| Follow graph | Graph store (edge lists in sharded SQL or Cassandra, plus cache) | Two adjacency lists per user; fast "followers of X" for fan-out. See Social Graph. |
| Home timelines, counters | Redis (backed by Cassandra) | Sub-millisecond reads for the hottest path in the app. |
| Direct messages | Cassandra / HBase, partitioned by conversation | Append-heavy, time-ordered per conversation. |
| Search documents | Elasticsearch | Inverted index, fuzzy and prefix matching; rebuilt from events. |
| Photos, videos | Object storage + CDN | Large immutable blobs; never in a database row. |
Rough numbers (assumptions)
Assume 500M daily active users; 100M new posts/day; each post gets an average of 40 likes and 2 comments; each active user generates 10 likes/day.
Likes: 500M × 10 = 5 B likes/day ≈ 58,000 writes/s average, ~150,000/s peak
Posts: 100M / 86,400 ≈ 1,160 posts/s
Feed fan-out: 1,160 posts/s × ~200 followers each ≈ 232,000 timeline inserts/s
Users table: 2 B rows × ~1 KB ≈ 2 TB → fits a few dozen PostgreSQL shards comfortably
Likes rows: 5 B/day × ~40 B ≈ 200 GB/day ≈ 73 TB/year → wide-column territory
The numbers tell the story: the user table is "small" and correctness-critical (SQL); likes and timelines are enormous and append-heavy (NoSQL). See Back-of-the-Envelope Estimation.
14. Trade-offs & common pitfalls
✅ Relational (SQL) strengths
- ACID transactions and enforced constraints
- Flexible ad-hoc queries and joins
- Mature tooling, well-known by every engineer
- Normalised data — one source of truth per fact
⚠️ NoSQL strengths (and SQL's weaknesses)
- Horizontal scaling of writes is built in, not bolted on
- Huge write throughput (LSM storage), multi-DC replication
- Flexible schemas for evolving data
- …but no joins, limited transactions, query-first modelling, eventual consistency to reason about
Common pitfalls
- Choosing NoSQL "for scale" at small scale. A single well-tuned PostgreSQL with replicas handles tens of thousands of transactions per second. Premature NoSQL costs you transactions and flexibility for nothing.
- Modelling Cassandra like SQL — normalised tables, then discovering you need
ALLOW FILTERING(a full-cluster scan) for every real query. - Unbounded partitions — partitioning by
post_idalone for likes on a post that gets 50M likes. Bucket it. - Read-modify-write in application code at read committed → lost updates. Use atomic updates,
SELECT … FOR UPDATEor version checks. - Assuming the default isolation is serializable. It almost never is (PostgreSQL: read committed; MySQL: repeatable read).
- Long transactions holding locks or old MVCC snapshots → lock waits, table bloat, replication lag.
- Storing blobs in the database. Images belong in object storage; store only the key.
- Dual writes without a plan — writing to PostgreSQL and Elasticsearch from the app, one fails, they diverge forever. Use events/CDC from a single source of truth.
- Queue-like tables in Cassandra (insert, process, delete) → tombstone accumulation and slow reads.
15. Real-world technologies
| Technology | Type | When to use it |
|---|---|---|
| PostgreSQL | Relational | Default choice for most products; rich SQL, JSONB, extensions (PostGIS, Citus). Instagram's original and core store. |
| MySQL (InnoDB) | Relational | Battle-tested OLTP at huge scale (Facebook, YouTube via Vitess, Shopify). |
| Cassandra / ScyllaDB | Wide-column | Massive write rates, time-ordered data per key, multi-region active-active. ScyllaDB is a C++ re-implementation for lower latency. |
| HBase / Bigtable | Wide-column | Strongly consistent per-row, Hadoop/GCP ecosystems, messaging and analytics backends. |
| DynamoDB | Key-value / document | Fully managed, predictable single-digit-ms latency, pay-per-request on AWS. |
| MongoDB | Document | Aggregate-oriented apps with evolving schemas; built-in sharding. |
| Redis / Memcached | Key-value (in-memory) | Caching, counters, leaderboards, rate limiting, timelines. |
| Neo4j / Neptune / JanusGraph | Graph | Multi-hop traversals, recommendations, fraud detection. |
| Prometheus / InfluxDB / TimescaleDB | Time-series | Metrics, IoT, monitoring. |
| Elasticsearch / OpenSearch | Search | Full-text and fuzzy search, log analytics — as a derived index. |
| Spanner / CockroachDB / TiDB / YugabyteDB | NewSQL | Relational + strong consistency beyond one machine or across regions. |
16. Interview corner
Which database(s) would you use for Instagram, and why?
Polyglot. Users and account data in sharded PostgreSQL: modest size (a few TB), needs uniqueness (usernames, emails) and transactions. Posts, comments, likes, activity and timelines in Cassandra: billions of writes per day, read by "entity, newest first", a natural partition-key/clustering-key fit, multi-region replication, eventual consistency acceptable. Hot timelines and counters in Redis. Search in Elasticsearch fed by Kafka events. Media in object storage behind a CDN. Each store has one source of truth; derived stores are rebuilt from events. This matches what Instagram has publicly described (sharded PostgreSQL + Cassandra).
Explain the isolation levels and the anomalies each prevents.
Read uncommitted allows dirty reads. Read committed prevents dirty reads (you only see committed data) but a row can change between two reads in your transaction (non-repeatable read) and new rows can appear in a range (phantom). Repeatable read gives each transaction a stable snapshot, preventing non-repeatable reads; the SQL standard still allows phantoms, though snapshot-based engines like PostgreSQL prevent them. Serializable guarantees the outcome equals some serial execution, preventing all of these plus lost updates and write skew. Lost updates aren't formally in the standard's table: PostgreSQL's repeatable read detects them, MySQL's doesn't. In practice most apps use read committed and handle hot spots with atomic updates, SELECT FOR UPDATE, or optimistic version checks.
Design a Cassandra table for "show the latest comments on a post". What about a post with 10 million comments?
PRIMARY KEY ((post_id), comment_id) WITH CLUSTERING ORDER BY (comment_id DESC), where comment_id is a time-sortable ID. The query WHERE post_id = ? LIMIT 20 is a single-partition sequential read. For 10M comments (~2 GB) the partition is far too large and hot, so bucket it: PRIMARY KEY ((post_id, bucket), comment_id) where bucket is a time window (day) or comment_id / N. Read the newest bucket first and walk backwards. Optionally keep the top/most-relevant comments in a cache.
How would you guarantee usernames are unique in a sharded or NoSQL system?
Uniqueness is a global invariant, so it needs a single authority per value. Options: (1) a dedicated usernames table sharded by username (not user_id) with a primary key on the username, so the uniqueness check is a single-shard insert that fails on conflict; (2) in Cassandra, INSERT INTO usernames … IF NOT EXISTS (a Paxos lightweight transaction) — slow but rare; (3) a NewSQL/consensus store. Then create the user record, and treat the claim as a reservation that is released if sign-up fails (compensation instead of 2PC).
Normalize or denormalize the feed?
Denormalize. The feed is the most read path in the app and must render in tens of ms. Store per-user timeline entries (post_id, author_id, maybe author name/avatar and media key) so one read returns everything needed, avoiding cross-shard joins. Writes cost more (fan-out on post), and copies like the author's avatar can go stale; fix by keeping only IDs in the timeline and hydrating hot objects (posts, users) from a cache, or by propagating change events. Keep the normalised source of truth elsewhere.
When would you choose a NewSQL database like Spanner or CockroachDB instead of sharded PostgreSQL?
When you need multi-row ACID transactions or strongly consistent secondary indexes across what would be multiple shards, data larger than one machine, and you'd rather not build and operate a sharding layer — e.g. a global user directory, billing, inventory. You pay higher write latency (consensus round trips, possibly cross-region), higher cost, and a less mature ecosystem. For simple, shard-local access patterns (everything keyed by user_id), app-level sharding of PostgreSQL is cheaper and faster.
What does BASE mean, and is Cassandra "eventually consistent"?
Basically Available, Soft state, Eventual consistency: keep serving under failures and let replicas converge later. Cassandra is tunably consistent: with consistency level ONE it's eventually consistent; with QUORUM writes and QUORUM reads (R + W > N) a read always overlaps the latest successful write, giving strong per-key read-after-write. It still lacks multi-partition transactions, and conflicts are resolved by last-write-wins timestamps.
17. Check your understanding
Two requests try to register @alice at the same instant on a single PostgreSQL. What prevents both from succeeding, and which ACID letter is that?
The UNIQUE constraint on username, backed by a unique index. One insert commits; the other waits on the index entry and then fails with a unique-violation. Enforcing declared invariants is the C (consistency) in ACID, made safe under concurrency by I (isolation).
At read committed, a job does SELECT like_count, adds 1 in Python, then UPDATE … SET like_count = 101. What anomaly can occur and give two fixes.
A lost update: two concurrent jobs read 100 and both write 101. Fixes: an atomic in-database increment SET like_count = like_count + 1; SELECT … FOR UPDATE to lock the row; or optimistic concurrency with a version column and retry.
In PRIMARY KEY ((user_id, day), activity_id), what are the partition key and clustering key, and can you query "all activity of type 'like' for user 42"?
Partition key: (user_id, day); clustering key: activity_id. Querying by type isn't efficient — type is neither in the partition key nor a clustering column, so it would require filtering every row of every day bucket. If that's a real query, create another table such as activity_by_user_type ((user_id, type, day), activity_id).
A celebrity receives 3 million likes per day, each like row ~60 bytes. Is likes_by_post ((post_id), user_id) safe for a single viral post over a week?
3M × 7 × 60 B ≈ 1.26 GB in one partition — about 10× the commonly cited ~100 MB guideline, and all its traffic lands on the same 3 replicas. Not safe. Bucket the partition (e.g. (post_id, bucket) with bucket = hash(user_id) % 16 for membership checks), and serve the count from a separate counter/cache rather than counting rows.
Which isolation level is needed to prevent write skew, and what's an example in Instagram?
Serializable (or explicit locks on the rows the decision depends on). Example: a group chat must always keep at least one admin; two admins simultaneously check "there are 2 admins" and both demote themselves, leaving zero. Snapshot isolation allows this because they wrote different rows.
Match each to a NoSQL family: (a) "accounts similar to selna gomz", (b) API p99 latency per minute, (c) a user's session token, (d) friends-of-friends who follow a hashtag.
(a) search engine (fuzzy matching over an inverted index); (b) time-series; (c) key-value (Redis/DynamoDB, with TTL); (d) graph (multi-hop traversal).
Why does the CQL schema store the follow edge in two tables, while the SQL schema uses one table plus an index?
In SQL, a secondary index on (followee_id, follower_id) answers "who follows me?" from the same table, maintained transactionally by the database. In Cassandra each table can only be queried efficiently by its own partition key, so "following of X" (partition = follower) and "followers of X" (partition = followee) need separate tables written together — classic query-first denormalisation.
Why might a durable COMMIT still lose data after a disk failure, and what fixes it?
Single-node durability only protects against crashes and power loss (WAL fsync'd to that disk). If the disk or machine dies, the data is gone unless the WAL was also shipped to another machine. Synchronous or semi-synchronous replication (and backups) extend durability across machines.
- Relational databases give you schemas, joins, constraints and ACID; they scale reads with replicas but writes only via sharding.
- Know the anomalies — dirty read, non-repeatable read, phantom, lost update, write skew — and which isolation level stops each. Defaults are weaker than serializable.
- Normalise for correctness, denormalise hot read paths and keep copies in sync with events.
- NoSQL = six families (key-value, document, wide-column, graph, time-series, search); pick by access pattern, scale and consistency needs.
- Cassandra modelling is query-first: partition key = where, clustering key = order; one table per query; bound partition size with buckets.
- Instagram is polyglot: sharded PostgreSQL for the relational core, Cassandra for high-write feeds/activity, plus Redis, Elasticsearch and object storage.
Related lessons
- Database Indexing & Storage Engines — B-trees vs LSM-trees: why SQL reads and Cassandra writes are fast
- Replication — read replicas, quorums (R + W > N) and durability across machines
- Sharding — how Instagram splits PostgreSQL into thousands of logical shards
- Consistent Hashing — how Cassandra maps partition keys to nodes
- CAP Theorem & Consistency Models — the theory behind ACID vs BASE
- Unique ID Generation — Instagram's 64-bit time-sortable IDs used as clustering keys
- Caching — Redis in front of the databases for hot reads
- Likes, Comments & Counters — handling hot counters without lost updates