Databases

CREATE TABLE users (
    user_id         BIGINT PRIMARY KEY AUTO_INCREMENT,
    username        VARCHAR(30) UNIQUE NOT NULL,
    email           VARCHAR(255) UNIQUE NOT NULL,
    password_hash   VARCHAR(255) NOT NULL,
    full_name       VARCHAR(100),
    bio             TEXT,
    profile_pic_url VARCHAR(500),
    follower_count  INTEGER DEFAULT 0,
    following_count INTEGER DEFAULT 0,
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                    ON UPDATE CURRENT_TIMESTAMP
);

Mistake 1: Using INTEGER for IDs on a large-scale system
INTEGER max: 2,147,483,647 (~2.1 billion)
Instagram has 2 billion users. You'd run out of IDs.
Always use BIGINT for systems that could reach billions of records.

Mistake 2: Storing passwords as plaintext
password VARCHAR(255)  -- WRONG
password_hash VARCHAR(255)  -- CORRECT (bcrypt, argon2)

Mistake 3: Storing images in the database as BLOBs
profile_pic BLOB  -- WRONG (database not designed for large binary data)
profile_pic_url VARCHAR(500)  -- CORRECT (store URL, image lives in S3)

-- Single column primary key
CREATE TABLE users (
    user_id BIGINT PRIMARY KEY,  -- Each user has a unique user_id
    username VARCHAR(30)
);

-- Composite primary key (two columns together form the key)
CREATE TABLE follows (
    follower_id BIGINT,
    followee_id BIGINT,
    PRIMARY KEY (follower_id, followee_id)  -- The combination must be unique
);
-- This means: User A can follow User B exactly once.
-- The pair (A, B) can exist only once in the table.

CREATE TABLE posts (
    post_id  BIGINT PRIMARY KEY,
    user_id  BIGINT NOT NULL,
    caption  TEXT,

    FOREIGN KEY (user_id) REFERENCES users(user_id)
        ON DELETE CASCADE
);

-- This enforces:
-- 1. Every post must belong to a valid user (user_id must exist in users table)
-- 2. ON DELETE CASCADE: If a user is deleted, all their posts are deleted too
-- 3. You CANNOT insert a post with user_id = 999 if user 999 doesn't exist

Without foreign keys:
- Application code must manually check if user exists before creating a post
- If user is deleted, orphaned posts remain in the database forever
- Data integrity depends on every developer writing correct code

With foreign keys:
- Database enforces the relationship automatically
- Impossible to create invalid data, regardless of application bugs
- Cascading deletes keep data consistent

Trade-off at scale:
Some systems at extreme scale (billions of writes/day) disable foreign key
checks for performance. The application layer handles consistency instead.
This is a deliberate trade-off, not a default choice.

-- INNER JOIN: Only rows that match in BOTH tables
SELECT users.username, posts.caption
FROM users
INNER JOIN posts ON users.user_id = posts.user_id;
-- Returns: Only users who have posts

-- LEFT JOIN: All rows from left table, matched rows from right (NULL if no match)
SELECT users.username, posts.caption
FROM users
LEFT JOIN posts ON users.user_id = posts.user_id;
-- Returns: ALL users, even those with no posts (caption = NULL for them)

-- RIGHT JOIN: All rows from right table, matched rows from left
-- (Less commonly used, you can usually rewrite as LEFT JOIN)

-- FULL OUTER JOIN: All rows from both tables, NULLs where no match

-- Get feed for user 123: recent posts from users they follow,
-- with author info and whether user 123 has liked each post

SELECT
    p.post_id,
    p.caption,
    p.image_url,
    p.created_at,
    p.like_count,
    p.comment_count,
    u.username,
    u.profile_pic_url,
    CASE WHEN l.user_id IS NOT NULL THEN true ELSE false END AS liked_by_me
FROM follows f
INNER JOIN posts p ON f.followee_id = p.user_id
INNER JOIN users u ON p.user_id = u.user_id
LEFT JOIN likes l ON p.post_id = l.post_id AND l.user_id = 123
WHERE f.follower_id = 123
ORDER BY p.created_at DESC
LIMIT 20;

Without index:  Read all 500 pages, note every mention.
                Time: proportional to total pages (O(n))

With index:     Look up "TCP" in the back-of-book index.
                Index says: pages 34, 89, 156, 312
                Go directly to those pages.
                Time: proportional to log of total pages (O(log n))

-- Without index: Full table scan
SELECT * FROM users WHERE email = 'mukul@example.com';
-- Database reads ALL rows (2 billion for Instagram)
-- Time: 10-60 seconds at scale

-- Create an index on the email column
CREATE INDEX idx_email ON users(email);

-- With index: Index lookup
SELECT * FROM users WHERE email = 'mukul@example.com';
-- Database looks up email in B-tree index, finds row pointer
-- Time: <1 millisecond

Table size vs lookup steps (B-tree with branching factor 100):

Rows in table          Index levels    Lookup time
1,000                  2 levels        <0.1ms
1,000,000              3 levels        <0.5ms
1,000,000,000          5 levels        <1ms

Even with 1 billion rows, a B-tree index lookup touches only
5 nodes. Each node is one disk read. Total: 5 disk reads.

Compare to full table scan: 1 billion rows = millions of disk reads.

-- Composite index on (user_id, created_at)
CREATE INDEX idx_user_posts ON posts(user_id, created_at DESC);

-- This single index efficiently serves:
SELECT * FROM posts WHERE user_id = 123 ORDER BY created_at DESC;

-- The index first narrows to user_id = 123,
-- then the results are already sorted by created_at DESC.
-- No additional sorting step required.

-- PostgreSQL: EXPLAIN shows the query execution plan
EXPLAIN ANALYZE SELECT * FROM users WHERE email = 'mukul@example.com';

-- Without index:
-- Seq Scan on users  (cost=0.00..450000.00 rows=1 width=200)
-- Actual time: 28543.121ms
-- ^^^^^^^^^ Full table scan, 28 seconds

-- With index:
-- Index Scan using idx_email on users  (cost=0.43..8.45 rows=1 width=200)
-- Actual time: 0.043ms
-- ^^^^^^^^^ Index lookup, 0.04 milliseconds

INSERT INTO posts (user_id, caption, created_at) VALUES (123, 'Hello', NOW());

Without indexes: 1 write operation (insert row into table)
With 3 indexes:  4 write operations (insert row + update 3 index structures)

Each index adds roughly 10-30% overhead to write operations.
3 indexes ≈ 30-90% slower writes.

Index columns that appear in:
- WHERE clauses (filtering)
- JOIN conditions (combining tables)
- ORDER BY clauses (sorting)
- Frequently accessed lookups (user by email, post by user)

Do NOT index:
- Columns rarely used in queries
- Columns with very low cardinality (e.g., boolean: only true/false)
- Tables that are almost exclusively write-heavy (logs, events)
- Every column "just in case" (waste of storage and write performance)

Atomicity:
  Server crashes after deducting from A but before adding to B.
  Without atomicity: A lost $500, B got nothing. Money vanished.
  With atomicity: Entire transaction rolled back. A still has $1000, B has $200.

Consistency:
  Business rule: Total money in system must remain constant ($1200).
  Before: A=$1000 + B=$200 = $1200
  After:  A=$500  + B=$700 = $1200  ✓
  If any constraint would be violated, transaction is rejected.

Isolation:
  While this transfer is running, another transaction reads Account A's balance.
  Without isolation: Could read $500 (mid-transaction) or $1000 (pre-transaction).
  With isolation: Reads either $1000 (before commit) or $500 (after commit), never a partial state.

Durability:
  Transaction commits successfully at 10:00:00.
  Server crashes at 10:00:01.
  When server restarts: A=$500, B=$700. The committed data is preserved.
  Implementation: Write-Ahead Log (WAL). Database writes to log before applying to data files.
  On crash recovery, it replays the log.

Redis (in-memory key-value store):
- GET/SET latency: ~0.1-0.5 milliseconds
- Throughput: 100,000-500,000 operations per second (single node)
- Data stored in RAM (fast) with optional disk persistence

DynamoDB (persistent key-value store):
- GET/SET latency: ~1-5 milliseconds
- Throughput: virtually unlimited (auto-scaling)
- Data stored on SSD with replication

// Users collection
{
    "_id": "user_123",
    "username": "mukul_oxford",
    "email": "mukul@example.com",
    "profile": {
        "full_name": "Mukul Raina",
        "bio": "Software Engineer",
        "location": "Toronto, Canada"
    },
    "interests": ["system-design", "distributed-systems", "machine-learning"],
    "follower_count": 10500
}

// Posts collection
{
    "_id": "post_789",
    "author_id": "user_123",
    "caption": "Beautiful day in Toronto",
    "media": {
        "type": "image",
        "url": "https://cdn.example.com/img/789.jpg",
        "thumbnails": {
            "small": "https://cdn.example.com/img/789_150.jpg",
            "medium": "https://cdn.example.com/img/789_640.jpg"
        }
    },
    "tags": ["toronto", "photography"],
    "comments": [
        {"user_id": "user_456", "text": "Great shot!", "created_at": "..."},
        {"user_id": "user_789", "text": "Love this", "created_at": "..."}
    ]
}

Misconception 1: "NoSQL is faster than SQL"

Reality: Speed depends on the access pattern, not the database type.
- Key lookup: Redis (~0.1ms) is faster than PostgreSQL (~1ms)
- Complex join across 4 tables: PostgreSQL handles it, MongoDB cannot
- The right tool for the right job. Neither is universally faster.

Misconception 2: "NoSQL scales better than SQL"

Reality: NoSQL databases were designed for horizontal scaling from
the start. SQL databases can also scale horizontally via sharding,
but it requires manual configuration.
- Cassandra: Add a node, it auto-rebalances. Simple.
- PostgreSQL: Set up sharding logic, shard router, manage cross-shard queries. Complex.
Both can reach the same scale. The operational effort differs.

Misconception 3: "Pick one or the other"

Reality: Most production systems use both.
Instagram uses:
- PostgreSQL for relational data (users, posts, follows)
- Redis for caching (feed cache, session store)
- Cassandra for high-volume writes (analytics events)

This is called "polyglot persistence" and is the standard
approach in modern system design.

Misconception 4: "NoSQL means no schema"

Reality: NoSQL means no schema enforced by the database.
Your application code still needs to understand the data structure.
The schema moves from database-level enforcement to application-level
enforcement. This can be a benefit (flexibility) or a risk
(inconsistent data if application code has bugs).

-- Customers table (customer data stored ONCE)
CREATE TABLE customers (
    customer_id BIGINT PRIMARY KEY,
    name        VARCHAR(100),
    email       VARCHAR(255) UNIQUE
);

-- Products table (product data stored ONCE)
CREATE TABLE products (
    product_id BIGINT PRIMARY KEY,
    name       VARCHAR(100),
    price      DECIMAL(10,2)
);

-- Orders table (only relationships and order-specific data)
CREATE TABLE orders (
    order_id    BIGINT PRIMARY KEY,
    customer_id BIGINT REFERENCES customers(customer_id),
    product_id  BIGINT REFERENCES products(product_id),
    quantity    INTEGER,
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Now: Alice's email exists in exactly one row.
-- Updating it means updating one row, not hunting through the entire database.

Normalized query: "How many followers does user 123 have?"

SELECT COUNT(*) FROM follows WHERE followee_id = 123;

If user 123 has 10 million followers, this query scans 10 million rows.
Even with an index, COUNT(*) over millions of rows takes hundreds of milliseconds.
On a profile page loaded millions of times per day, this is unacceptable.

Denormalized approach: Store the count directly in the users table.

SELECT follower_count FROM users WHERE user_id = 123;

Single row lookup, <1ms. The count is pre-computed and stored redundantly.

Write path:
1. Application sends INSERT/UPDATE/DELETE to primary
2. Primary executes the query
3. Primary writes change to WAL (Write-Ahead Log)
4. Primary sends WAL entries to all replicas (asynchronously)
5. Each replica applies the changes to its local copy

Read path:
1. Application sends SELECT to one of the replicas (round-robin)
2. Replica executes query against its local copy
3. Returns result to application

Read distribution:
- Primary: handles all writes + ~10% of reads (read-after-write cases)
- Replica 1: ~30% of reads
- Replica 2: ~30% of reads
- Replica 3: ~30% of reads

Total read capacity: 4x a single database

Per-shard configuration:
- 1 Primary: Handles all writes
- 3 Read Replicas: Handle reads

Capacity per shard:
- Write: up to 10,000 writes/sec (primary only)
- Read: up to 40,000 reads/sec (primary + 3 replicas)

Cost:
- 4 database instances per shard
- ~$500/month per instance = $2,000/month per shard

def create_post(user_id, image, caption):
    # Write to primary
    post_id = primary_db.insert(posts, {...})

    # Mark this user as "recently wrote" in cache
    cache.set(f"recent_write:{user_id}", True, ttl=5)  # 5-second window

    return post_id

def get_user_posts(user_id):
    # Check if this user wrote something in the last 5 seconds
    if cache.get(f"recent_write:{user_id}"):
        # Read from primary (guaranteed to have latest data)
        return primary_db.query("SELECT * FROM posts WHERE user_id = ?", user_id)
    else:
        # Safe to read from replica (no recent writes)
        return replica_db.query("SELECT * FROM posts WHERE user_id = ?", user_id)

Single PostgreSQL instance practical limits:
- Storage: ~10-16 TB
- Write throughput: ~10,000 writes/sec
- Connection limit: ~5,000 concurrent connections

Instagram's data:
- 36 TB metadata per year (growing)
- Historical data accumulates
- 500M daily active users generating queries

A single database simply cannot hold or serve all this data.
Sharding distributes the data across multiple machines.

Given:
- 2 billion users
- 36 TB metadata per year
- 100,000 read QPS, 1,000 write QPS

Shard configuration:
- Number of shards: 1,000
- Shard key: user_id % 1000
- Users per shard: 2,000,000 (2 billion / 1,000)
- Data per shard: ~36 GB (36 TB / 1,000)

Each shard is a PostgreSQL instance with:
- 1 Primary + 3 Read Replicas = 4 instances per shard
- Total instances: 1,000 × 4 = 4,000 database instances

Read capacity per shard:
- 4 instances × 10,000 reads/sec = 40,000 reads/sec
- Total: 1,000 shards × 40,000 = 40,000,000 reads/sec capacity
- Required: 100,000 reads/sec
- Headroom: 400x (room for growth and traffic spikes)

Write capacity per shard:
- 1 primary × 10,000 writes/sec = 10,000 writes/sec
- Total: 1,000 shards × 10,000 = 10,000,000 writes/sec capacity
- Required: 1,000 writes/sec
- Headroom: 10,000x

Problem: Celebrity user has 600 million followers.
Shard 345 (where the celebrity lives) gets disproportionate load
whenever the celebrity posts and millions of users request their content.

Solutions:
- Dedicated shard for celebrity accounts
- Read replicas specifically for hot shards
- Cache celebrity data aggressively in Redis
- Fan-out strategy (covered in Course 1: hybrid fan-out)

When individual shards grow too large, you need to increase shard count.

From 1,000 shards to 2,000 shards:
- user_id % 1000 = 345   (old shard)
- user_id % 2000 = 345 or 1345   (new shard assignment)
- ~50% of data must be migrated

This is a complex, risky operation. Consistent hashing reduces the
amount of data that moves during re-sharding (covered in Section 11).

1. "I'll just use a database" without specifying which type or why
2. Designing schema without indexes
3. Using NoSQL "because it scales better" without justifying the access pattern
4. Not mentioning replication or sharding when designing for millions of users
5. Storing images or files in the database instead of object storage
6. Using arrays in SQL columns instead of junction tables
7. Not considering cross-shard query implications when choosing a shard key

- B-tree index lookup: O(log n), <1ms even for billions of rows
- Full table scan: O(n), seconds to minutes at scale
- Single PostgreSQL instance: ~10,000 QPS, ~10-16 TB storage limit
- Redis GET/SET: ~0.1-0.5ms, 100K-500K ops/sec per node
- Replication lag: 100-500ms (asynchronous)
- Shard capacity: ~2 million users per shard (at Instagram's scale)