Antwa CodeAntwaCode Blog

Cari

Rekayasa7 menit baca

Database Design Patterns yang Wajib Diketahui Developer

Pattern database design yang penting untuk developer.

Baca dalam English →

Database Design Patterns yang Wajib Diketahui Developer

Pernah nggak sih, aplikasi kamu yang tadinya lancar jaya tiba-tiba jadi lemot parah pas user-nya naik dari 100 ke 10.000? Atau mungkin schema database kamu udah kayak spaghetti — saling ketergantungan, susah di-maintain, dan tiap kali ada perubahan kecil harus bongkar habis?

Kalau iya, kemungkinan besar masalahnya ada di database design. Dan bukan sekadar "bikin tabel, bikin kolom" — tapi tentang pattern atau pola desain yang tepat untuk kondisi tertentu.

Di artikel ini, kita bakal bedah 7 database design pattern yang wajib banget diketahui developer. Kita mulai dari yang fundamental sampai yang advanced. Siap? Let's go! 🚀

Table of Contents

  1. Normalisasi (Normalization)
  2. Strategi Indexing
  3. Caching Patterns
  4. Sharding
  5. Read Replicas
  6. Event Sourcing
  7. CQRS (Command Query Responsibility Segregation)
  8. Kapan Harus Pakai Pattern Apa?
  9. Kesimpulan

Normalisasi (Normalization)

Normalisasi adalah fondasi dari database relational. Intinya: hilangkan data redundan, pastikan setiap fakta disimpan di satu tempat saja.

Kenapa Penting?

Bayangkan kamu punya tabel orders yang nyimpen nama customer di setiap baris order. Kalau "Budi Santoso" punya 100 order, nama "Budi Santoso" ditulis 100 kali. Kalau dia ganti nama jadi "Budi Prasetyo", kamu harus update 100 baris. That's a maintenance nightmare.

Tingkatan Normalisasi

First Normal Form (1NF) — Setiap kolom hanya berisi satu nilai (atomic), tidak ada kolom yang nyimpen list atau array.

-- ❌ VIOLASI 1NF
CREATE TABLE orders_bad (
    id INT PRIMARY KEY,
    customer_name VARCHAR(100),
    items TEXT  -- "Laptop,Mouse,Keyboard" — ini harus dipecah!
);
 
-- ✅ MEMATUHI 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) — Sudah 1NF, dan setiap kolom non-key bergantung pada seluruh primary key (bukan sebagian).

Ini relevan kalau pakai composite primary key:

-- ❌ VIOLASI 2NF
CREATE TABLE order_details (
    order_id INT,
    product_id INT,
    product_name VARCHAR(100),  -- Bergantung ke product_id saja, bukan composite key
    quantity INT,
    PRIMARY KEY (order_id, product_id)
);
 
-- ✅ MEMATUHI 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) — Sudah 2NF, dan tidak ada transitive dependency (kolom non-key tidak bergantung ke kolom non-key lainnya).

-- ❌ VIOLASI 3NF
CREATE TABLE employees (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    department_id INT,
    department_name VARCHAR(100)  -- Transitive: name → department_id → department_name
);
 
-- ✅ MEMATUHI 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) — Stricter version dari 3NF. Setiap determinan harus superkey. Dalam praktik sehari-hari, 3NF sudah cukup untuk kebanyakan kasus.

Kapan Deliberately Denormalize?

Tapi tunggu — normalisasi bukan berarti "selalu benar." Dalam kasus tertentu, denormalisasi justru lebih baik:

  • Read-heavy applications: Kalau kamu 90% baca dan 10% tulis, denormalisasi untuk mempercepat query
  • Data warehousing / analytics: JOIN yang kompleks di tabel ratusan juta baris itu mahal
  • Caching layer: Materialized view atau pre-computed columns
-- Denormalisasi untuk dashboard: 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: Normalisasi dulu, denormalize kalau ada bottleneck yang terukur.

Strategi Indexing

Kalau normalisasi itu soal struktur, indexing itu soal kecepatan akses. Tanpa index yang tepat, query kamu bisa jadi full table scan — artinya database harus baca SELURUH tabel cuma untuk cari satu baris.

B-Tree Index (Default)

Ini adalah index paling umum di hampir semua relational database (PostgreSQL, MySQL InnoDB). Cocok untuk:

  • Range queries (BETWEEN, >, <, >=, <=)
  • Sorting (ORDER BY)
  • Equality checks (=)
-- Index untuk query yang sering filter by email
CREATE INDEX idx_users_email ON users(email);
 
-- Query ini sekarang pakai index, bukan full scan
SELECT * FROM users WHERE email = 'budi@example.com';
 
-- Range query juga pakai index
SELECT * FROM users WHERE created_at BETWEEN '2024-01-01' AND '2024-12-31';

Hash Index

Cocok untuk equality checks saja — lebih cepat dari B-tree untuk lookup langsung, tapi tidak support range queries.

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

Kapan pakai: tabel lookup yang hanya perlu exact match, seperti session storage atau cache key.

Composite Index (Multi-column)

Index yang mencakup beberapa kolom sekaligus. Urutan kolom matters!

-- Index untuk query yang filter by status DAN sort by date
CREATE INDEX idx_orders_status_date ON orders(status, created_at);
 
-- ✅ Pakai index (kolom pertama cocok)
SELECT * FROM orders WHERE status = 'pending' ORDER BY created_at DESC;
 
-- ❌ GAGAL pakai index (skip kolom pertama)
SELECT * FROM orders WHERE created_at > '2024-01-01';

Rule of thumb: Letakkan kolom yang paling sering di-WHERE di posisi pertama, diikuti kolom ORDER BY.

Covering Index

Index yang mencakup semua kolom yang dibutuhkan query. Database tidak perlu kembali ke tabel asli (tidak perlu "bookmark lookup").

-- Covering index untuk dashboard query
CREATE INDEX idx_orders_covering 
    ON orders(status, created_at, total_amount);
 
-- Query ini sepenuhnya dilayani dari index
SELECT status, created_at, total_amount 
FROM orders 
WHERE status = 'completed'
ORDER BY created_at DESC;

Untuk mengecek apakah query pakai covering index di PostgreSQL:

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

Partial Index

Index yang hanya mencakup subset baris. Hemat storage dan maintenance.

-- Index hanya untuk order yang belum selesai
CREATE INDEX idx_pending_orders 
    ON orders(created_at) 
    WHERE status = 'pending';
 
-- Query ini pakai partial index
SELECT * FROM orders WHERE status = 'pending' AND created_at > '2024-01-01';

Anti-patterns Indexing

Hati-hati dengan beberapa kesalahan umum:

-- ❌ Terlalu banyak index → lambat saat INSERT/UPDATE
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 index lagi → setiap INSERT harus update 20 index!
 
-- ❌ Index pada kolom dengan low cardinality
CREATE INDEX idx_gender ON users(gender);
-- Gender hanya punya 2-3 nilai → index tidak efektif
 
-- ❌ Over-indexing kolom yang jarang di-query
-- Selalu ukur dengan EXPLAIN ANALYZE sebelum buat index

Tip: Pakai pg_stat_user_indexes (PostgreSQL) untuk cek index yang tidak pernah dipakai:

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

Caching Patterns

Database itu lambat (relatif). Kalau setiap request harus ke database, throughput-nya terbatas. Caching adalah pola untuk menyimpan data di tempat yang lebih cepat (Redis, Memcached, in-memory) supaya tidak perlu query database berulang kali.

Cache-Aside (Lazy Loading)

Ini adalah pattern paling populer. Aplikasi yang manage cache-nya sendiri.

Request → Cek Cache → Cache HIT? → Return data
                       Cache MISS? → Query DB → Simpan ke 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: Cek cache dulu
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)  # Cache HIT!
    
    # Step 2: Cache MISS → query database
    user = db.query("SELECT * FROM users WHERE id = %s", [user_id])
    
    # Step 3: Simpan ke cache (TTL 1 jam)
    redis_client.setex(cache_key, 3600, json.dumps(user))
    
    return user

Kelebihan: Hanya cache data yang benar-benar diminta (lazy). Kekurangan: Cache miss pertama selalu kena database.

Write-Through

Setiap write ke database juga langsung write ke cache. Data selalu sinkron.

Write → Update DB → Update Cache → Return
def update_user(user_id: int, data: dict) -> dict:
    # Step 1: Update database
    db.execute("UPDATE users SET name = %s WHERE id = %s", 
               [data['name'], user_id])
    
    # Step 2: Update 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

Kelebihan: Data di cache selalu fresh. Kekurangan: Write latency lebih tinggi (harus tulis ke 2 tempat).

Write-Behind (Write-Back)

Write ke cache dulu, lalu asynchronously ke database. Bagus untuk 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 ke cache dulu (cepat!)
    cache_key = f"user:{user_id}"
    redis_client.setex(cache_key, 3600, json.dumps(data))
    
    # Queue write ke database (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()
 
# Jalankan di background thread
threading.Thread(target=db_writer, daemon=True).start()

Kelebihan: Write latency sangat rendah. Kekurangan: Bisa ada data loss kalau cache crash sebelum ke-database. Gunakan dengan persistent queue!

Cache Invalidation Strategies

"Invalidation is one of the two hardest things in computer science" — dan itu benar.

TTL-based: Set expiry time, data expired akan dihapus otomatis.

# TTL 1 jam untuk data yang jarang berubah
redis_client.setex("product:123", 3600, json.dumps(product))
 
# TTL 5 menit untuk data yang sering berubah (price, stock)
redis_client.setex("product:123:price", 300, json.dumps(price))

Event-based: Invalidate cache saat data berubah.

# Subscribe ke database change events (PostgreSQL LISTEN/NOTIFY, MySQL binlog)
def on_order_update(order_id):
    # Hapus cache terkait
    redis_client.delete(f"order:{order_id}")
    redis_client.delete(f"user:{order.user_id}:orders")  # Cache list juga

Versioned keys: Tambahkan versi ke 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 adalah teknik memecah satu database besar menjadi beberapa database yang lebih kecil (shard), masing-masing menyimpan subset data.

Kenapa Butuh Sharding?

  • Satu database sudah mentok (tidak bisa scale vertikal lagi)
  • Query makin lambat karena tabel terlalu besar
  • Butuh low latency untuk user di berbagai region

Hash-Based Sharding

Distribusi data berdasarkan hash dari shard key.

import hashlib
 
NUM_SHARDS = 4
 
def get_shard(user_id: int) -> int:
    """Tentukan shard berdasarkan 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
# dst...
 
def get_db_connection(user_id: int):
    shard = get_shard(user_id)
    return connections[shard]  # connections = [db_shard_0, db_shard_1, ...]

Kelebihan: Distribusi seragam, simple. Kekurangan: Tidak bisa query lintas shard (cross-shard queries mahal), sulit tambah shard baru (rebalancing).

Range-Based Sharding

Berdasarkan range nilai tertentu.

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)
 
# Data regional terkumpul di shard yang sama → queries regional cepat!

Kelebihan: Queries regional sangat cepat (data lokal). Kekurangan: Hotspot (shard Indonesia bisa jauh lebih besar dari shard Thailand).

Consistent Hashing

Solusi untuk masalah rebalancing saat tambah/hapus shard.

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]]
 
# Tambah/hapus shard hanya mempengaruhi tetangga terdekat
ring = ConsistentHashRing(['shard_0', 'shard_1', 'shard_2'])
shard = ring.get_node("user:12345")

Horizontal vs Vertical Sharding

AspekHorizontalVertical
PemecahanBaris (data)Kolom (fitur)
ContohUser A di shard 1, User B di shard 2Profile di DB 1, Logs di DB 2
KompleksitasTinggi (cross-shard)Sedang (join lintas DB)
Cocok untukScaling data/userScaling fitur yang berbeda

Read Replicas

Read Replicas adalah pola di mana kamu punya satu database primary untuk write dan beberapa replica untuk read. Distribusi beban baca ke beberapa instance.

Arsitektur Dasar

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

Implementasi dengan 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 balancing ke replica (round-robin atau random)"""
        return random.choice(self.replicas)
    
    def get_write_connection(self):
        """Selalu ke 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 → ke replica
users = db.query("SELECT * FROM users WHERE active = true")
 
# Write → ke primary
db.query("INSERT INTO users (name) VALUES (%s)", ["Budi"])

Replication Lag Problem

Masalah terbesar read replicas: replication lag. Setelah write ke primary, data butuh waktu untuk propagate ke replica (biasanya milidetik sampai detik).

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)
        # Tandai bahwa user ini baru saja di-write
        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')):
            # Cek apakah ini read-after-write pattern
            if params and params[0] in self._recent_writes:
                elapsed = time.time() - self._recent_writes[params[0]]
                if elapsed < 5:  # 5 detik terakhir, baca dari primary
                    return self.primary.execute(sql, params)
            return random.choice(self.replicas).execute(sql, params)
        return self.primary.execute(sql, params)

Cara mengatasi replication lag:

  1. Read-your-writes consistency: Setelah write, baca dari primary untuk beberapa detik
  2. Causal consistency: Gunakan timestamp atau version untuk tahu apakah replica sudah up-to-date
  3. Synchronous replication: Lebih konsisten, tapi lebih lambat

Kapan Pakai Read Replicas?

  • Application dengan read-to-write ratio tinggi (misal 10:1 atau lebih)
  • Butuh high availability (kalau primary down, bisa promote replica)
  • Need geographic distribution (replica di region berbeda untuk latency rendah)

Event Sourcing

Event Sourcing adalah pola di mana kamu tidak menyimpan state terkini dari suatu entity, tapi menyimpan serangkaian event (perubahan) yang menghasilkan state tersebut.

Konsep Dasar

Bayangkan kamu sedang membangun aplikasi financial ledger. Daripada hanya menyimpan saldo akhir, kamu simpan setiap transaksi:

Traditional approach:

-- State terkini saja
CREATE TABLE accounts (
    id INT PRIMARY KEY,
    balance DECIMAL(15,2)
);
-- balance = 5000000  ← tapi gimana bisa segitu? Transaksi apa aja?

Event Sourcing:

-- Semua event yang pernah terjadi
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
);
 
-- Contoh 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       |

Untuk mendapatkan state terkini, replay semua event:

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

Replay event dari awal itu lambat kalau event-nya sudah jutaan. Snapshot menyimpan state pada titik tertentu:

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:
    # Ambil snapshot terakhir
    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 event SETELAH 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)
    
    # Simpan snapshot baru (setiap 100 event)
    if state.version - start_version > 100:
        save_snapshot(account_id, state)
    
    return state

Event Store

Untuk production, pertimbangkan 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 dengan 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 semua event sekaligus
        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]
        )

Kapan Event Sourcing?

  • Audit trail wajib: Financial, healthcare, legal — perlu tahu siapa ubah apa kapan
  • Temporal queries: "Apa state order pada tanggal 15 Maret?" — replay event sampai tanggal itu
  • Debugging: Replay event untuk mereproduksi bug
  • Integration: Event bisa jadi input untuk sistem lain

CQRS (Command Query Responsibility Segregation)

CQRS adalah pola yang memisahkan model untuk write (command) dan model untuk read (query). Kombinasikan dengan Event Sourcing untuk hasil yang powerful.

Konsep Dasar

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

Implementasi Sederhana

# ===== 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. Validasi & simpan ke 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  # Database terpisah, denormalized
    
    def get_order_detail(self, order_id: int) -> dict:
        # Query dari 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 yang update read model"""
    
    def __init__(self, read_db):
        self.db = read_db
    
    def on_order_created(self, event):
        # Upsert ke 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

Ketika digabung, kamu dapat:

  • Full audit trail dari event sourcing
  • Read-optimized views dari CQRS
  • Independent scaling — read DB bisa scale horizontal, write DB bisa scale vertical
  • Flexible read models — buat view yang berbeda untuk kebutuhan yang berbeda tanpa mengubah write model

Kapan Pakai CQRS?

  • Microservices dengan command dan query workload yang sangat berbeda
  • Domain yang kompleks dengan banyak operasi baca yang berbeda
  • Performance kritis — butuh read latency rendah sambil menulis ke database yang heavy
  • Bilingual/Audit requirement — perlu history lengkap dari semua perubahan

Kapan Harus Pakai Pattern Apa?

Ini cheat sheet cepat berdasarkan masalah yang kamu hadapi:

MasalahPatternKapan
Data redundan, update susahNormalisasiSelalu mulai dari sini
Query lambatIndexingUkur dulu dengan EXPLAIN ANALYZE
Database sering kecapaianCachingBaca > tulis, data tidak terlalu kritis
Satu DB mentokShardingSudah vertical scale maksimal, butuh write throughput
Read bottleneckRead ReplicasRead:write ratio > 5:1
Perlu audit trail lengkapEvent SourcingFinancial, legal, compliance
Read & write workload berbedaCQRSMicroservices, complex domains

Kombinasi yang Sering Digunakan

Startup kecil (0-10k users):
  Normalisasi + Indexing + Cache-Aside
 
Menengah (10k-1M users):
  + Read Replicas + Partial Denormalization
 
Besar (1M+ users):
  + Sharding + CQRS + Event Sourcing

Anti-pattern: Over-engineering

Jangan langsung pakai Event Sourcing + CQRS + Sharding untuk aplikasi todo list yang cuma dipake 5 orang. Mulai dari yang sederhana, tambah kompleksitas hanya saat ada bottleneck yang terukur.

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

Kesimpulan

Database design pattern bukan ilmu yang dipelajari sekali lalu dilupakan. Ini adalah toolbox yang terus berkembang seiring pertumbuhan aplikasi kamu. Beberapa poin penting:

  1. Mulai dari normalisasi — ini fondasinya. Denormalize saat ada bottleneck, bukan karena malas bikin JOIN.

  2. Index secara strategis — jangan over-index, jangan under-index. Selalu pakai EXPLAIN ANALYZE untuk verifikasi.

  3. Caching adalah senjata ampuh — tapi cache invalidation itu harder than it looks. Pilih strategi yang sesuai dengan tolerance kamu terhadap stale data.

  4. Scaling horizontal ada harganya — Sharding dan read replicas menambah kompleksitas. Pastikan kamu benar-benar butuh sebelum adopt.

  5. Event Sourcing dan CQRS untuk domain kompleks — powerful, tapi learning curve-nya tinggi. Pakai hanya kalau memang dibutuhkan.

  6. Ukur sebelum optimasi — Jangan asumsikan, ukur. EXPLAIN ANALYZE, profiling query, monitoring — itu yang menentukan pattern mana yang harus dipakai.

Semoga artikel ini jadi referensi yang berguna saat kamu desain atau redesign database. Selamat menulis SQL yang efisien! 💾

Artikel ini ditulis untuk developer Indonesia yang ingin memahami database design patterns secara mendalam. Kalau ada pertanyaan atau mau diskusi lebih lanjut, jangan ragu untuk komen di bawah!

Tulisan lain