EngineeringAugust 17, 20268 min read

Database Design Patterns Every Developer Should Know

Important database design patterns for developers.

Database Design Patterns Every Developer Should Know

Ever had an app that ran perfectly smooth suddenly crawl to a halt when users jumped from 100 to 10,000? Or maybe your database schema has become spaghetti — tangled dependencies, impossible to maintain, and every minor change requires tearing everything apart?

If so, the problem is almost certainly database design. And not just "create a table, add columns" — but about choosing the right design patterns for your specific situation.

In this article, we'll break down 7 database design patterns that every developer absolutely needs to know. We'll start from fundamentals and work our way to advanced techniques. Ready? Let's go! 🚀

Normalization
Indexing Strategies
Caching Patterns
Sharding
Read Replicas
Event Sourcing
CQRS (Command Query Responsibility Segregation)
When to Use Which Pattern?
Conclusion

Normalization

Normalization is the foundation of relational databases. The core idea: eliminate redundant data and make sure every fact is stored in exactly one place.

Why Does It Matter?

Imagine you have an orders table that stores the customer name in every order row. If "Budi Santoso" has 100 orders, the name "Budi Santoso" appears 100 times. If he changes his name to "Budi Prasetyo," you'd have to update 100 rows. That's a maintenance nightmare.

Normal Forms

First Normal Form (1NF) — Every column contains a single value (atomic). No columns should store lists or arrays.

-- ❌ VIOLATES 1NF
CREATE TABLE orders_bad (
    id INT PRIMARY KEY,
    customer_name VARCHAR(100),
    items TEXT  -- "Laptop,Mouse,Keyboard" — needs to be split!
);
 
-- ✅ COMPLIES WITH 1NF
CREATE TABLE orders (
    id INT PRIMARY KEY,
    customer_id INT,
    order_date DATE
);
 
CREATE TABLE order_items (
    id INT PRIMARY KEY,
    order_id INT,
    product_name VARCHAR(100),
    quantity INT
);

Second Normal Form (2NF) — Already in 1NF, and every non-key column depends on the entire primary key (not just part of it).

This matters when you use composite primary keys:

-- ❌ VIOLATES 2NF
CREATE TABLE order_details (
    order_id INT,
    product_id INT,
    product_name VARCHAR(100),  -- Depends on product_id alone, not the composite key
    quantity INT,
    PRIMARY KEY (order_id, product_id)
);
 
-- ✅ COMPLIES WITH 2NF
CREATE TABLE order_details (
    order_id INT,
    product_id INT,
    quantity INT,
    PRIMARY KEY (order_id, product_id),
    FOREIGN KEY (product_id) REFERENCES products(id)
);

Third Normal Form (3NF) — Already in 2NF, and no transitive dependencies exist (non-key columns don't depend on other non-key columns).

-- ❌ VIOLATES 3NF
CREATE TABLE employees (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    department_id INT,
    department_name VARCHAR(100)  -- Transitive: name → department_id → department_name
);
 
-- ✅ COMPLIES WITH 3NF
CREATE TABLE employees (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    department_id INT,
    FOREIGN KEY (department_id) REFERENCES departments(id)
);
 
CREATE TABLE departments (
    id INT PRIMARY KEY,
    name VARCHAR(100)
);

Boyce-Codd Normal Form (BCNF) — A stricter version of 3NF. Every determinant must be a superkey. In practice, 3NF is sufficient for most use cases.

When Should You Deliberately Denormalize?

But wait — normalization isn't always the right answer. In certain cases, denormalization actually performs better:

Read-heavy applications: If you're 90% reads and 10% writes, denormalize to speed up queries
Data warehousing / analytics: Complex JOINs on tables with hundreds of millions of rows are expensive
Caching layer: Materialized views or pre-computed columns

-- Denormalization for dashboards: pre-compute total spend
CREATE TABLE customer_summary (
    customer_id INT PRIMARY KEY,
    customer_name VARCHAR(100),
    total_orders INT DEFAULT 0,
    total_spent DECIMAL(15,2) DEFAULT 0,
    last_order_date DATE,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Rule of thumb: Normalize first, denormalize only when there's a measurable bottleneck.

Indexing Strategies

While normalization is about structure, indexing is about access speed. Without the right indexes, your queries can turn into full table scans — meaning the database has to read the entire table just to find a single row.

B-Tree Index (The Default)

This is the most common index type in nearly every relational database (PostgreSQL, MySQL InnoDB). It's great for:

Range queries (BETWEEN, >, <, >=, <=)
Sorting (ORDER BY)
Equality checks (=)

-- Index for frequently filtering by email
CREATE INDEX idx_users_email ON users(email);
 
-- This query now uses the index instead of a full scan
SELECT * FROM users WHERE email = 'budi@example.com';
 
-- Range queries also use the index
SELECT * FROM users WHERE created_at BETWEEN '2024-01-01' AND '2024-12-31';

Hash Index

Best for equality checks only — faster than B-tree for direct lookups, but doesn't support range queries.

-- PostgreSQL syntax
CREATE INDEX idx_users_email_hash ON users USING hash(email);

When to use: Lookup tables that only need exact matches, like session storage or cache keys.

Composite Index (Multi-column)

An index that covers multiple columns at once. Column order matters!

-- Index for queries that filter by status AND sort by date
CREATE INDEX idx_orders_status_date ON orders(status, created_at);
 
-- ✅ Uses the index (first column matches)
SELECT * FROM orders WHERE status = 'pending' ORDER BY created_at DESC;
 
-- ❌ DOESN'T use the index (skips the first column)
SELECT * FROM orders WHERE created_at > '2024-01-01';

Rule of thumb: Put the most frequently used WHERE columns first, followed by ORDER BY columns.

Covering Index

An index that includes every column the query needs. The database never has to go back to the original table (no "bookmark lookup" needed).

-- Covering index for dashboard queries
CREATE INDEX idx_orders_covering 
    ON orders(status, created_at, total_amount);
 
-- This query is fully served from the index
SELECT status, created_at, total_amount 
FROM orders 
WHERE status = 'completed'
ORDER BY created_at DESC;

To verify whether a query uses a covering index in PostgreSQL:

EXPLAIN (ANALYZE, BUFFERS) 
SELECT status, created_at, total_amount 
FROM orders 
WHERE status = 'completed' 
ORDER BY created_at DESC;
-- Look for "Index Only Scan" in the output

Partial Index

An index that only covers a subset of rows. Saves storage and maintenance overhead.

-- Index only for pending orders
CREATE INDEX idx_pending_orders 
    ON orders(created_at) 
    WHERE status = 'pending';
 
-- This query uses the partial index
SELECT * FROM orders WHERE status = 'pending' AND created_at > '2024-01-01';

Indexing Anti-patterns

Watch out for these common mistakes:

-- ❌ Too many indexes → slow INSERTs/UPDATEs
CREATE INDEX idx_users_a ON users(col_a);
CREATE INDEX idx_users_b ON users(col_b);
CREATE INDEX idx_users_c ON users(col_c);
-- ... 20 more indexes → every INSERT must update 20 indexes!
 
-- ❌ Index on a low-cardinality column
CREATE INDEX idx_gender ON users(gender);
-- Gender has only 2-3 values → index isn't effective
 
-- ❌ Over-indexing columns that are rarely queried
-- Always verify with EXPLAIN ANALYZE before creating an index

Tip: Use pg_stat_user_indexes (PostgreSQL) to find indexes that are never used:

SELECT indexrelname, idx_scan 
FROM pg_stat_user_indexes 
WHERE idx_scan = 0 
ORDER BY pg_relation_size(indexrelid) DESC;

Caching Patterns

Databases are slow (relatively). If every request has to hit the database, your throughput is limited. Caching is the pattern of storing data in a faster location (Redis, Memcached, in-memory) so you don't have to query the database repeatedly.

Cache-Aside (Lazy Loading)

This is the most popular pattern. The application manages its own cache.

Request → Check Cache → Cache HIT? → Return data
                        Cache MISS? → Query DB → Store in cache → Return data

import redis
import json
 
redis_client = redis.Redis(host='localhost', port=6379, db=0)
 
def get_user(user_id: int) -> dict:
    cache_key = f"user:{user_id}"
    
    # Step 1: Check cache first
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)  # Cache HIT!
    
    # Step 2: Cache MISS → query the database
    user = db.query("SELECT * FROM users WHERE id = %s", [user_id])
    
    # Step 3: Store in cache (TTL: 1 hour)
    redis_client.setex(cache_key, 3600, json.dumps(user))
    
    return user

Pros: Only caches data that's actually requested (lazy). Cons: The first cache miss always hits the database.

Write-Through

Every write to the database also writes directly to the cache. Data is always in sync.

Write → Update DB → Update Cache → Return

def update_user(user_id: int, data: dict) -> dict:
    # Step 1: Update the database
    db.execute("UPDATE users SET name = %s WHERE id = %s", 
               [data['name'], user_id])
    
    # Step 2: Update the cache
    cache_key = f"user:{user_id}"
    updated_user = db.query("SELECT * FROM users WHERE id = %s", [user_id])
    redis_client.setex(cache_key, 3600, json.dumps(updated_user))
    
    return updated_user

Pros: Cache data is always fresh. Cons: Higher write latency (must write to two places).

Write-Behind (Write-Back)

Writes go to the cache first, then asynchronously to the database. Great for write-heavy workloads.

Write → Update Cache → Return → (async) → Update DB

import threading
import queue
 
write_queue = queue.Queue()
 
def update_user_write_behind(user_id: int, data: dict) -> dict:
    # Write to cache first (fast!)
    cache_key = f"user:{user_id}"
    redis_client.setex(cache_key, 3600, json.dumps(data))
    
    # Queue the database write (async)
    write_queue.put(('update_user', user_id, data))
    
    return data
 
# Background worker
def db_writer():
    while True:
        task_type, user_id, data = write_queue.get()
        if task_type == 'update_user':
            db.execute("UPDATE users SET name = %s WHERE id = %s",
                      [data['name'], user_id])
        write_queue.task_done()
 
# Run in a background thread
threading.Thread(target=db_writer, daemon=True).start()

Pros: Very low write latency. Cons: Data loss is possible if the cache crashes before syncing to the database. Use a persistent queue!

Cache Invalidation Strategies

"Invalidation is one of the two hardest things in computer science" — and it's true.

TTL-based: Set an expiry time; expired data gets deleted automatically.

# 1-hour TTL for data that rarely changes
redis_client.setex("product:123", 3600, json.dumps(product))
 
# 5-minute TTL for frequently changing data (price, stock)
redis_client.setex("product:123:price", 300, json.dumps(price))

Event-based: Invalidate the cache when data changes.

# Subscribe to database change events (PostgreSQL LISTEN/NOTIFY, MySQL binlog)
def on_order_update(order_id):
    # Clear related cache entries
    redis_client.delete(f"order:{order_id}")
    redis_client.delete(f"user:{order.user_id}:orders")  # Clear the order list cache too

Versioned keys: Add a version number to the cache key.

def get_user_v2(user_id: int) -> dict:
    version = redis_client.get(f"user:{user_id}:version") or "1"
    cache_key = f"user:{user_id}:v{version}"
    # ...

Sharding

Sharding is the technique of splitting one large database into multiple smaller databases (shards), each storing a subset of the data.

Why Do You Need Sharding?

A single database has hit its ceiling (can't scale vertically anymore)
Queries are getting slower because tables are too large
You need low latency for users across multiple regions

Hash-Based Sharding

Distributes data based on a hash of the shard key.

import hashlib
 
NUM_SHARDS = 4
 
def get_shard(user_id: int) -> int:
    """Determine the shard based on user_id"""
    hash_val = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16)
    return hash_val % NUM_SHARDS
 
# User 1 → shard 1
# User 2 → shard 3
# User 3 → shard 0
# etc...
 
def get_db_connection(user_id: int):
    shard = get_shard(user_id)
    return connections[shard]  # connections = [db_shard_0, db_shard_1, ...]

Pros: Uniform distribution, simple to implement. Cons: Cross-shard queries are expensive; adding new shards requires rebalancing.

Range-Based Sharding

Splits data based on a value range.

def get_shard_by_region(user_id: int, region: str) -> int:
    region_shards = {
        'ID': 0,  # Indonesia → shard 0
        'MY': 1,  # Malaysia → shard 1
        'SG': 2,  # Singapore → shard 2
        'TH': 3,  # Thailand → shard 3
    }
    return region_shards.get(region, 0)
 
# Regional data stays on the same shard → fast regional queries!

Pros: Regional queries are very fast (data is local). Cons: Hotspots (the Indonesia shard could become much larger than the Thailand shard).

Consistent Hashing

Solves the rebalancing problem when adding or removing shards.

import bisect
import hashlib
 
class ConsistentHashRing:
    def __init__(self, nodes, virtual_nodes=150):
        self.ring = {}
        self.sorted_keys = []
        for node in nodes:
            for i in range(virtual_nodes):
                key = self._hash(f"{node}:{i}")
                self.ring[key] = node
                bisect.insort(self.sorted_keys, key)
    
    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def get_node(self, data_key):
        h = self._hash(data_key)
        idx = bisect.bisect_right(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0
        return self.ring[self.sorted_keys[idx]]
 
# Adding/removing shards only affects the nearest neighbors
ring = ConsistentHashRing(['shard_0', 'shard_1', 'shard_2'])
shard = ring.get_node("user:12345")

Horizontal vs Vertical Sharding

Aspect	Horizontal	Vertical
Split by	Rows (data)	Columns (features)
Example	User A on shard 1, User B on shard 2	Profiles in DB 1, Logs in DB 2
Complexity	High (cross-shard)	Medium (cross-DB joins)
Best for	Scaling data/users	Scaling different features

Read Replicas

Read Replicas are a pattern where you have one primary database for writes and multiple replicas for reads. Read load is distributed across several instances.

Basic Architecture

                    ┌─────────────────┐
                    │   Application   │
                    └────────┬────────┘
                             │
                    ┌────────▼────────┐
                    │  Load Balancer   │
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              │              │              │
     ┌────────▼───────┐ ┌───▼────────┐ ┌──▼──────────┐
     │  Primary (RW)  │ │ Replica 1  │ │ Replica 2   │
     │  10.0.0.1      │ │ (RO)       │ │ (RO)        │
     └────────────────┘ └────────────┘ └─────────────┘
           │                    ▲              ▲
           │    replication     │              │
           └────────────────────┴──────────────┘

Implementation with Connection Routing

import random
 
class DatabaseRouter:
    def __init__(self):
        self.primary = create_connection("10.0.0.1:5432")
        self.replicas = [
            create_connection("10.0.0.2:5432"),
            create_connection("10.0.0.3:5432"),
            create_connection("10.0.0.4:5432"),
        ]
        self.write_lock = False
    
    def get_read_connection(self):
        """Load balance across replicas (round-robin or random)"""
        return random.choice(self.replicas)
    
    def get_write_connection(self):
        """Always use the primary"""
        return self.primary
    
    def query(self, sql, params=None):
        if sql.strip().upper().startswith(('SELECT', 'WITH')):
            conn = self.get_read_connection()
        else:
            conn = self.get_write_connection()
        return conn.execute(sql, params)
 
# Usage
db = DatabaseRouter()
 
# Read → to a replica
users = db.query("SELECT * FROM users WHERE active = true")
 
# Write → to the primary
db.query("INSERT INTO users (name) VALUES (%s)", ["Budi"])

The Replication Lag Problem

The biggest challenge with read replicas: replication lag. After writing to the primary, data takes time to propagate to replicas (usually milliseconds to seconds).

class ReadAfterWriteRouter(DatabaseRouter):
    def __init__(self):
        super().__init__()
        self._recent_writes = {}  # user_id → timestamp
    
    def write(self, sql, params=None):
        result = self.primary.execute(sql, params)
        # Mark this user as recently written
        if params and 'user_id' in str(sql):
            self._recent_writes[params[0]] = time.time()
        return result
    
    def query(self, sql, params=None):
        if sql.strip().upper().startswith(('SELECT', 'WITH')):
            # Check for read-after-write pattern
            if params and params[0] in self._recent_writes:
                elapsed = time.time() - self._recent_writes[params[0]]
                if elapsed < 5:  # Written within last 5 seconds, read from primary
                    return self.primary.execute(sql, params)
            return random.choice(self.replicas).execute(sql, params)
        return self.primary.execute(sql, params)

How to handle replication lag:

Read-your-writes consistency: After a write, read from the primary for a few seconds
Causal consistency: Use timestamps or versions to determine if a replica is up-to-date
Synchronous replication: More consistent, but slower

When to Use Read Replicas?

Applications with a high read-to-write ratio (e.g., 10:1 or more)
You need high availability (if the primary goes down, promote a replica)
Geographic distribution is needed (replicas in different regions for lower latency)

Event Sourcing

Event Sourcing is a pattern where you don't store the current state of an entity — instead, you store a series of events (changes) that produced that state.

Core Concept

Imagine you're building a financial ledger. Rather than just storing the final balance, you store every single transaction:

Traditional approach:

-- Only the current state
CREATE TABLE accounts (
    id INT PRIMARY KEY,
    balance DECIMAL(15,2)
);
-- balance = 5000000  ← but how did it get there? What transactions happened?

Event Sourcing:

-- Every event that ever occurred
CREATE TABLE account_events (
    id BIGSERIAL PRIMARY KEY,
    account_id INT NOT NULL,
    event_type VARCHAR(50) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMP DEFAULT NOW(),
    version INT NOT NULL
);
 
-- Example data
-- | account_id | event_type      | payload                              | version |
-- |------------|-----------------|--------------------------------------|---------|
-- | 1          | AccountCreated  | {"initial_balance": 0}               | 1       |
-- | 1          | MoneyDeposited  | {"amount": 5000000, "ref": "TRX001"} | 2       |
-- | 1          | MoneyWithdrawn  | {"amount": 2000000, "ref": "TRX002"} | 3       |
-- | 1          | MoneyDeposited  | {"amount": 2000000, "ref": "TRX003"} | 4       |

To get the current state, replay all events:

from dataclasses import dataclass
from typing import List
 
@dataclass
class AccountState:
    balance: float = 0
    version: int = 0
 
def apply_event(state: AccountState, event: dict) -> AccountState:
    if event['event_type'] == 'AccountCreated':
        state.balance = event['payload']['initial_balance']
    elif event['event_type'] == 'MoneyDeposited':
        state.balance += event['payload']['amount']
    elif event['event_type'] == 'MoneyWithdrawn':
        state.balance -= event['payload']['amount']
    state.version = event['version']
    return state
 
def get_account_state(account_id: int) -> AccountState:
    events = db.query(
        "SELECT * FROM account_events WHERE account_id = %s ORDER BY version",
        [account_id]
    )
    state = AccountState()
    for event in events:
        state = apply_event(state, event)
    return state
 
# Result: AccountState(balance=5000000, version=4) ✅

Snapshot Pattern

Replaying events from the beginning gets slow once you have millions of them. Snapshots save state at a specific point:

CREATE TABLE account_snapshots (
    account_id INT PRIMARY KEY,
    balance DECIMAL(15,2),
    version INT NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

def get_account_state_optimized(account_id: int) -> AccountState:
    # Fetch the most recent snapshot
    snapshot = db.query_one(
        "SELECT * FROM account_snapshots WHERE account_id = %s ORDER BY version DESC LIMIT 1",
        [account_id]
    )
    
    state = AccountState()
    start_version = 0
    
    if snapshot:
        state.balance = snapshot['balance']
        state.version = snapshot['version']
        start_version = snapshot['version']
    
    # Replay events AFTER the snapshot
    events = db.query(
        "SELECT * FROM account_events WHERE account_id = %s AND version > %s ORDER BY version",
        [account_id, start_version]
    )
    
    for event in events:
        state = apply_event(state, event)
    
    # Save a new snapshot (every 100 events)
    if state.version - start_version > 100:
        save_snapshot(account_id, state)
    
    return state

Event Store

For production use, consider a dedicated event store:

class EventStore:
    def __init__(self, db_connection):
        self.db = db_connection
        self.subscribers = []
    
    def append(self, aggregate_id: str, events: List[dict]):
        """Append events with optimistic concurrency"""
        current_version = self._get_current_version(aggregate_id)
        
        for event in events:
            current_version += 1
            event['version'] = current_version
            event['aggregate_id'] = aggregate_id
            event['timestamp'] = datetime.utcnow().isoformat()
        
        # Insert all events at once
        self.db.execute_many(
            """INSERT INTO events (aggregate_id, event_type, payload, version, timestamp)
               VALUES (%(aggregate_id)s, %(event_type)s, %(payload)s, %(version)s, %(timestamp)s)""",
            events
        )
        
        # Notify subscribers
        for event in events:
            for subscriber in self.subscribers:
                subscriber(event)
    
    def get_events(self, aggregate_id: str, after_version: int = 0) -> List[dict]:
        return self.db.query(
            """SELECT * FROM events 
               WHERE aggregate_id = %s AND version > %s 
               ORDER BY version""",
            [aggregate_id, after_version]
        )

When to Use Event Sourcing?

Audit trails are mandatory: Financial, healthcare, legal — you need to know who changed what and when
Temporal queries: "What was the order state on March 15?" — replay events up to that date
Debugging: Replay events to reproduce bugs
Integration: Events can serve as input for other systems

CQRS (Command Query Responsibility Segregation)

CQRS is a pattern that separates the write model (command) from the read model (query). Combine it with Event Sourcing for powerful results.

Core Concept

          ┌──────────────────────────────────┐
          │          Application             │
          └──────────┬───────────┬───────────┘
                     │           │
            ┌────────▼───┐  ┌───▼────────┐
            │  Command    │  │   Query    │
            │  (Write)    │  │   (Read)   │
            └────────┬────┘  └───┬────────┘
                     │           │
            ┌────────▼───┐  ┌───▼────────┐
            │ Write DB   │  │  Read DB   │
            │ (Normalize)│  │ (Denormal) │
            └────────────┘  └────────────┘

Simple Implementation

# ===== COMMAND SIDE (Write) =====
 
class CreateOrderCommand:
    def __init__(self, user_id: int, items: list, total: float):
        self.user_id = user_id
        self.items = items
        self.total = total
 
class OrderCommandHandler:
    def __init__(self, write_db, event_store):
        self.db = write_db
        self.events = event_store
    
    def handle_create_order(self, command: CreateOrderCommand) -> int:
        # 1. Validate & save to database (normalized)
        order_id = self.db.execute(
            """INSERT INTO orders (user_id, status, total, created_at)
               VALUES (%s, 'pending', %s, NOW()) RETURNING id""",
            [command.user_id, command.total]
        )
        
        for item in command.items:
            self.db.execute(
                """INSERT INTO order_items (order_id, product_id, quantity, price)
                   VALUES (%s, %s, %s, %s)""",
                [order_id, item['product_id'], item['quantity'], item['price']]
            )
        
        # 2. Publish event
        self.events.append(f"order:{order_id}", [{
            'event_type': 'OrderCreated',
            'payload': {
                'order_id': order_id,
                'user_id': command.user_id,
                'total': command.total,
                'items': command.items
            }
        }])
        
        return order_id
 
# ===== QUERY SIDE (Read) =====
 
class OrderQueryService:
    def __init__(self, read_db):
        self.db = read_db  # Separate database, denormalized
    
    def get_order_detail(self, order_id: int) -> dict:
        # Query from the read model (denormalized, pre-joined)
        return self.db.query_one(
            """SELECT o.*, u.name as user_name, u.email,
                      json_agg(json_build_object(
                          'product_name', p.name,
                          'quantity', oi.quantity,
                          'price', oi.price
                      )) as items
               FROM orders o
               JOIN users u ON o.user_id = u.id
               JOIN order_items oi ON o.id = oi.order_id
               JOIN products p ON oi.product_id = p.id
               WHERE o.id = %s
               GROUP BY o.id, u.name, u.email""",
            [order_id]
        )
    
    def get_user_orders_summary(self, user_id: int) -> list:
        # Optimized for listing — denormalized table
        return self.db.query(
            """SELECT order_id, user_name, total, status, 
                      item_count, created_at
               FROM order_read_model
               WHERE user_id = %s
               ORDER BY created_at DESC
               LIMIT 20""",
            [user_id]
        )

Syncing Write → Read

class ReadModelProjector:
    """Event listener that updates the read model"""
    
    def __init__(self, read_db):
        self.db = read_db
    
    def on_order_created(self, event):
        # Upsert into the read model
        self.db.execute("""
            INSERT INTO order_read_model 
                (order_id, user_id, user_name, user_email, total, status, 
                 item_count, created_at)
            SELECT 
                %(order_id)s, o.user_id, u.name, u.email, o.total, o.status,
                (SELECT COUNT(*) FROM order_items WHERE order_id = %(order_id)s),
                o.created_at
            FROM orders o
            JOIN users u ON o.user_id = u.id
            WHERE o.id = %(order_id)s
            ON CONFLICT (order_id) DO UPDATE SET
                status = EXCLUDED.status,
                total = EXCLUDED.total
        """, {'order_id': event['payload']['order_id']})
    
    def on_order_completed(self, event):
        self.db.execute(
            "UPDATE order_read_model SET status = 'completed' WHERE order_id = %s",
            [event['payload']['order_id']]
        )

CQRS + Event Sourcing = Power Combo

When combined, you get:

Full audit trail from event sourcing
Read-optimized views from CQRS
Independent scaling — the read DB can scale horizontally while the write DB scales vertically
Flexible read models — create different views for different needs without changing the write model

When to Use CQRS?

Microservices where command and query workloads are very different
Complex domains with many different read operations
Performance-critical systems that need low read latency while writing to a heavy database
Bilingual/audit requirements — when you need a complete history of all changes

When to Use Which Pattern?

Here's a quick cheat sheet based on the problem you're facing:

Problem	Pattern	When
Redundant data, hard to update	Normalization	Always start here
Slow queries	Indexing	Measure first with `EXPLAIN ANALYZE`
Database overload	Caching	Reads > writes, data isn't too critical
Single DB hit its ceiling	Sharding	Already at max vertical scale, need write throughput
Read bottleneck	Read Replicas	Read:write ratio > 5:1
Need full audit trail	Event Sourcing	Financial, legal, compliance
Different read & write workloads	CQRS	Microservices, complex domains

Common Combinations

Small startup (0-10k users):
  Normalization + Indexing + Cache-Aside
 
Medium (10k-1M users):
  + Read Replicas + Partial Denormalization
 
Large (1M+ users):
  + Sharding + CQRS + Event Sourcing

Anti-pattern: Over-engineering

Don't jump straight to Event Sourcing + CQRS + Sharding for a to-do list app used by 5 people. Start simple, add complexity only when there's a measurable bottleneck.

The best pattern is the simplest one that solves your actual problem.

Conclusion

Database design patterns aren't something you learn once and forget. They're a toolbox that evolves as your application grows. Here are the key takeaways:

Start with normalization — it's the foundation. Denormalize when you hit a bottleneck, not because you're too lazy to write JOINs.
Index strategically — don't over-index, don't under-index. Always verify with EXPLAIN ANALYZE.
Caching is a powerful weapon — but cache invalidation is harder than it looks. Choose a strategy that matches your tolerance for stale data.
Horizontal scaling comes at a price — Sharding and read replicas add complexity. Make sure you actually need them before adopting.
Event Sourcing and CQRS for complex domains — powerful, but with a steep learning curve. Only use them when truly needed.
Measure before optimizing — Don't assume, measure. EXPLAIN ANALYZE, query profiling, monitoring — that's what determines which pattern to use.

I hope this article serves as a useful reference when you're designing or redesigning your database. Happy SQL writing! 💾

This article was written for developers who want to understand database design patterns in depth. If you have questions or want to discuss further, feel free to leave a comment below!

CI/CD Pipeline with GitHub Actions: Complete Tutorial
August 19, 2026 · 8 min read
Docker for PHP/Laravel Developers: Complete Guide
August 19, 2026 · 3 min read
Microservices vs Monolith: Which One to Choose?
August 17, 2026 · 11 min read

Database Design Patterns Every Developer Should Know

Table of Contents

Normalization

Why Does It Matter?

Normal Forms

When Should You Deliberately Denormalize?

Indexing Strategies

B-Tree Index (The Default)

Hash Index

Composite Index (Multi-column)

Covering Index

Partial Index

Indexing Anti-patterns

Caching Patterns

Cache-Aside (Lazy Loading)

Write-Through

Write-Behind (Write-Back)

Cache Invalidation Strategies

Sharding

Why Do You Need Sharding?

Hash-Based Sharding

Range-Based Sharding

Consistent Hashing

Horizontal vs Vertical Sharding

Read Replicas

Basic Architecture

Implementation with Connection Routing

The Replication Lag Problem

When to Use Read Replicas?

Event Sourcing

Core Concept

Snapshot Pattern

Event Store

When to Use Event Sourcing?

CQRS (Command Query Responsibility Segregation)

Core Concept

Simple Implementation

Syncing Write → Read

CQRS + Event Sourcing = Power Combo

When to Use CQRS?

When to Use Which Pattern?

Common Combinations

Anti-pattern: Over-engineering

Conclusion

More posts

CI/CD Pipeline with GitHub Actions: Complete Tutorial

Docker for PHP/Laravel Developers: Complete Guide

Microservices vs Monolith: Which One to Choose?