Database Design & Selection

The five questions before you pick

1. Read vs write ratio — 90% reads favors replicas + cache; write-heavy favors LSM/wide-column.
2. Query shape — point lookup, range scan, join, aggregation, geo, full-text, traversal?
3. Consistency needs — financial ledger (strong) vs social likes (eventual)?
4. Scale trajectory — single node OK for 2 years, or 100M rows day one?
5. Operational maturity — managed service vs self-hosted Cassandra cluster?

flowchart TD
  start(["New data store needed"]) --> q1{"Structured schema\nwith joins?"}
  q1 -->|Yes| sql["Relational SQL\nPostgreSQL / MySQL"]
  q1 -->|No| q2{"Document-shaped\nnested JSON?"}
  q2 -->|Yes| doc["Document DB\nMongoDB / DynamoDB"]
  q2 -->|No| q3{"Massive write\nthroughput?"}
  q3 -->|Yes| wc["Wide-column\nCassandra / HBase"]
  q3 -->|No| q4{"Sub-ms latency\ncache/session?"}
  q4 -->|Yes| kv["Key-value\nRedis / Memcached"]
  q4 -->|No| q5{"Time-ordered\nmetrics/logs?"}
  q5 -->|Yes| ts["Time-series\nInfluxDB / TimescaleDB / ClickHouse"]
  q5 -->|No| q6{"Full-text or\nfaceted search?"}
  q6 -->|Yes| search["Search engine\nElasticsearch / OpenSearch"]
  q6 -->|No| q7{"Relationship\ntraversals?"}
  q7 -->|Yes| graphdb["Graph DB\nNeo4j / Neptune"]
  q7 -->|No| sql

Decision matrix by store family

Polyglot persistence — the norm at scale

Production systems rarely use one database. Uber runs MySQL (core trips), Cassandra (location), Redis (cache), Elasticsearch (search). The pattern: PostgreSQL as system of record, specialized stores as derived views. Sync via CDC (Debezium → Kafka) or dual-write (riskier). Interview signal: name the primary store and justify satellites.

ACID properties

• Atomicity — all statements in a transaction commit or none do (WAL + rollback segments).
• Consistency — constraints (FK, CHECK, UNIQUE) hold after every committed transaction.
• Isolation — concurrent transactions behave as if serialized (level-dependent).
• Durability — committed data survives crash (fsync WAL before ACK).

Isolation levels and anomalies

ANSI SQL defines four isolation levels. PostgreSQL implements three (Read Uncommitted behaves as Read Committed). Higher isolation = fewer anomalies, more locking/retries, lower throughput.

* PostgreSQL Repeatable Read prevents phantom reads for most cases via MVCC + predicate locking.

Indexes — B-tree, hash, GIN, partial

B-tree (default) supports equality and range on sortable columns. Composite index column order matters: (user_id, created_at) serves WHERE user_id = ? and WHERE user_id = ? AND created_at > ? but not WHERE created_at > ? alone.

• Covering index — INCLUDE columns avoid heap fetch (index-only scan).
• Partial index — WHERE status = 'active' shrinks index size 10× for hot subsets.
• GIN/GiST — JSONB, full-text, geo in PostgreSQL.
• Write amplification — every index doubles insert cost; over-indexing kills write-heavy tables.

-- Composite + partial index for active user orders
CREATE INDEX idx_orders_user_active
  ON orders (user_id, created_at DESC)
  WHERE status NOT IN ('cancelled', 'archived');

-- Covering index: avoid heap lookup for list view
CREATE INDEX idx_products_list
  ON products (category_id, price)
  INCLUDE (name, thumbnail_url);

-- Explain analyze before shipping
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, name, price FROM products
WHERE category_id = 42 AND price BETWEEN 10 AND 100
ORDER BY price LIMIT 20;

The N+1 query problem

Loading 100 posts then issuing 100 queries for authors = 101 round trips (~101 × 1ms = 101ms vs 2ms with JOIN). ORMs lazy-load by default—fix with eager loading, JOIN FETCH, or batch IN queries.

// BAD: N+1 — 1 query for posts + N for each author
List<Post> posts = postRepo.findByUserId(userId);
posts.forEach(p -> p.getAuthor().getName()); // lazy load each author

// GOOD: single JOIN or @EntityGraph
@Query("SELECT p FROM Post p JOIN FETCH p.author WHERE p.userId = :uid")
List<Post> findWithAuthor(@Param("uid") Long userId);

Connection pooling

PostgreSQL uses ~10 MB RAM per connection and a dedicated backend process. Opening a connection costs 1–5 ms. At 500 app instances × 20 connections = 10,000 backends — the database dies before your app does. PgBouncer (transaction pooling) multiplexes thousands of clients onto ~100 server connections.

• Pool size rule of thumb — connections = (core_count × 2) + effective_spindle_count (PgBouncer docs); often 20–50 per primary.
• Transaction vs session pooling — transaction mode breaks prepared statements and temp tables; session mode safer for ORMs.
• RDS Proxy / Aurora — managed pooling + IAM auth + failover.

Read replicas

Streaming replication ships WAL to standbys. Replicas serve read traffic; lag typically 10ms–seconds under load. Replica lag means read-your-writes is NOT guaranteed unless routed to primary or tracked LSN.

flowchart LR
  app[Application]
  primary[(Primary PostgreSQL)]
  replica1[(Read Replica 1)]
  replica2[(Read Replica 2)]
  app -->|writes + strong reads| primary
  app -->|analytics / lists| replica1
  app -->|search indexing| replica2
  primary -->|WAL stream| replica1
  primary -->|WAL stream| replica2

PostgreSQL vs MySQL

When to shard

• Single-node storage > 2–5 TB or write QPS exceeds ~10–50K sustained (hardware dependent).
• Replication lag unacceptable and writes are the bottleneck.
• Regulatory data residency requires geographic partitions.

Before sharding: archive cold data, denormalize, connection pool, read replicas, caching — each buys 10× cheaper than sharding.

Shard key selection

The shard key routes every query to the correct partition. Ideal properties:

• High cardinality — uniform distribution (user_id ✓, country_code ✗).
• Query alignment — most queries include shard key (avoid cross-shard scatter-gather).
• Immutability — changing shard key requires row migration.
• Monotonic keys avoided — auto-increment IDs concentrate writes on latest shard (use hash or snowflake IDs).

Sharding strategies

Consistent hashing

Plain mod N remaps almost all keys when adding a shard. Consistent hashing places shards and keys on a ring; adding a node only moves keys between adjacent neighbors (~K/N keys move). Used in DynamoDB, Cassandra, Redis Cluster, CDNs.

flowchart TB
  subgraph ring["Hash ring (virtual nodes)"]
    direction TB
    n1["Shard A\nvnodes 0-85"]
    n2["Shard B\nvnodes 86-170"]
    n3["Shard C\nvnodes 171-255"]
    n1 --> n2 --> n3 --> n1
  end
  key1["key: user_4821\n→ Shard B"]
  key2["key: user_9901\n→ Shard A"]

Virtual nodes (vnodes) — each physical shard owns many ring positions for even load after node failure. Typical: 256 vnodes per Cassandra node.

Cross-shard operations

Operations without shard key in the WHERE clause hit every shard:

• Scatter-gather — query all shards, merge in coordinator (latency = slowest shard).
• Global secondary indexes — index table sharded differently = double write + eventual consistency.
• Cross-shard transactions — 2PC is slow and fragile; prefer saga, per-shard transactions, or redesign.
• JOINs across shards — application-level join or denormalize; SQL JOINs don't span Vitess shards without careful schema.

Vitess — sharded MySQL at YouTube scale

Vitess sits between app and MySQL shards: VTGate routes SQL by shard key, VReplication handles resharding with minimal downtime, connection pooling built-in. Used by Slack, HubSpot, Square. Alternative: Citus for PostgreSQL (distributed tables + reference tables).

-- Vitess: colocated sharding schema
CREATE TABLE users (
  user_id BIGINT NOT NULL,
  email VARCHAR(255),
  PRIMARY KEY (user_id)
);

CREATE TABLE orders (
  order_id BIGINT NOT NULL,
  user_id BIGINT NOT NULL,
  total DECIMAL(10,2),
  PRIMARY KEY (order_id),
  KEY idx_user (user_id)
) VITESS_SHARDING_KEY = user_id;  -- same shard as users row

When documents win

• Heterogeneous records (CMS, catalogs, user profiles with optional fields).
• Read-heavy access patterns that fetch whole object graphs.
• Rapid schema evolution without ALTER TABLE migrations.
• Horizontal scale via sharded collections (MongoDB sharded cluster, DynamoDB partitions).

Embed vs reference

// EMBED: order with line items (bounded, always read together)
{
  "_id": ObjectId("..."),
  "userId": "u_4821",
  "items": [
    { "sku": "WIDGET-1", "qty": 2, "price": 19.99 },
    { "sku": "WIDGET-2", "qty": 1, "price": 49.99 }
  ],
  "total": 89.97,
  "status": "shipped"
}

// REFERENCE: product reviews (unbounded, paginated separately)
{
  "_id": ObjectId("..."),
  "productId": "prod_991",
  "rating": 5,
  "text": "Great widget",
  "userId": "u_100"
}

Aggregation pipeline

Server-side pipeline stages: $match → $group → $sort → $lookup (join). Push filtering early ($match first) to use indexes. Heavy analytics often belongs in warehouse (BigQuery) not OLTP Mongo.

// Top 10 products by revenue last 30 days
db.orders.aggregate([
  { $match: { createdAt: { $gte: ISODate("2026-05-08") }, status: "paid" } },
  { $unwind: "$items" },
  { $group: {
      _id: "$items.sku",
      revenue: { $sum: { $multiply: ["$items.qty", "$items.price"] } },
      units: { $sum: "$items.qty" }
  }},
  { $sort: { revenue: -1 } },
  { $limit: 10 }
]);

MongoDB sharding recap

• Shard key — immutable; choose compound key for locality (tenant_id + user_id).
• Chunk migration — balancer moves 64 MB chunks; jumbo chunks block balance.
• Transactions — multi-doc ACID within same shard since 4.0; cross-shard since 4.2 with performance cost.

Data model: partition key + cluster key

• Partition key — determines which node(s) store the row; all rows with same partition key live on same replica set.
• Cluster key (clustering columns) — sort order within partition; enables range scans inside one partition.
• Primary key = partition key + optional cluster key(s).

-- Messages by user: partition = user_id, cluster = conversation_id, sent_at
CREATE TABLE messages_by_user (
  user_id UUID,
  conversation_id UUID,
  sent_at TIMESTAMP,
  message_id UUID,
  body TEXT,
  PRIMARY KEY ((user_id), conversation_id, sent_at)
) WITH CLUSTERING ORDER BY (conversation_id ASC, sent_at DESC);

-- Query: all messages in one conversation for user (single partition)
SELECT * FROM messages_by_user
WHERE user_id = ? AND conversation_id = ? LIMIT 50;

flowchart LR
  subgraph node1["Node 1"]
    p1["Partition: user_A\nrows sorted by cluster key"]
  end
  subgraph node2["Node 2"]
    p2["Partition: user_B"]
  end
  subgraph node3["Node 3"]
    p3["Partition: user_C"]
  end
  router["Coordinator"] -->|hash partition key| node1
  router --> node2
  router --> node3

Tunable consistency

Cassandra trades latency vs staleness per query via ConsistencyLevel and replication factor (RF). QUORUM with RF=3 means 2 replicas must respond. LOCAL_QUORUM avoids cross-DC latency in multi-region.

Anti-patterns

• Hot partitions — celebrity user_id absorbs all writes for partition; salt key (user_id:bucket).
• ALLOW FILTERING — full table scan disguised as query; kills cluster at scale.
• Secondary indexes on high-cardinality columns — hidden scatter-gather; use materialized views or duplicate tables per query.
• Large partitions — >100 MB or millions of rows per partition causes compaction/repair pain; target <10 MB.
• Multi-DC strong consistency everywhere — WAN latency dominates; use LOCAL_QUORUM per DC.

HBase (brief)

HBase runs on HDFS — strong consistency per row, sorted by row key, built for billions of rows scanned in range. Write path through WAL + MemStore; reads merge StoreFiles. Used when Hadoop ecosystem already present (Facebook Messages history). Ops-heavy vs Cassandra's peer-to-peer simplicity; prefer Cassandra/Scylla for greenfield unless HDFS integration required.

Redis data structures

# Leaderboard: top 10 by score
ZADD leaderboard 9842.5 user:4821
ZADD leaderboard 9100.0 user:9901
ZREVRANGE leaderboard 0 9 WITHSCORES

# Rate limit: 100 requests per minute per IP
INCR ratelimit:192.168.1.1:202606071430
EXPIRE ratelimit:192.168.1.1:202606071430 60

# Cache-aside pattern (pseudo)
# val = GET key; if miss: val = db.query(); SET key val EX 300

Redis persistence

• RDB — point-in-time snapshots; fast restart; may lose last minutes of writes.
• AOF — append every write; appendfsync everysec typical; more durable, larger files.
• Neither — pure cache; acceptable if source of truth elsewhere.
• Redis Cluster — 16,384 hash slots across masters; replica failover; no multi-key ACID across slots.

flowchart TB
  client[Client]
  n1["Master 1\nslots 0-5460"]
  n2["Master 2\nslots 5461-10922"]
  n3["Master 3\nslots 10923-16383"]
  r1[Replica]
  r2[Replica]
  r3[Replica]
  client -->|CRC16 key → slot| n1
  client --> n2
  client --> n3
  n1 --> r1
  n2 --> r2
  n3 --> r3

DynamoDB

• Partition key — required; 10 GB per partition soft limit; hot keys throttle.
• Sort key — optional; enables range queries within partition.
• GSI (Global Secondary Index) — alternate access pattern with own partition key; eventual consistency; extra WCU/RCU.
• Streams — ordered change feed per shard; triggers Lambda, CDC to Elasticsearch, audit.
• On-demand vs provisioned — on-demand for spiky/unknown; provisioned + auto-scaling for predictable cost at scale.

{
  "TableName": "Orders",
  "KeySchema": [
    { "AttributeName": "userId", "KeyType": "HASH" },
    { "AttributeName": "orderId", "KeyType": "RANGE" }
  ],
  "GlobalSecondaryIndexes": [{
    "IndexName": "ByStatus",
    "KeySchema": [
      { "AttributeName": "status", "KeyType": "HASH" },
      { "AttributeName": "createdAt", "KeyType": "RANGE" }
    ],
    "Projection": { "ProjectionType": "INCLUDE", "NonKeyAttributes": ["total", "userId"] }
  }],
  "StreamSpecification": { "StreamViewType": "NEW_AND_OLD_IMAGES" }
}

Memcached vs Redis

Workload characteristics

• Write-heavy — millions of points/sec (Prometheus remote write, Influx line protocol).
• Time-range queries — last 1h, GROUP BY 1m buckets; rarely point lookups on old data.
• Retention tiers — raw 7d → 1min aggregates 90d → 1h aggregates 2y.
• Cardinality explosion — high-cardinality labels (user_id on every metric) OOM the index.

-- TimescaleDB hypertable + continuous aggregate
CREATE TABLE metrics (
  time        TIMESTAMPTZ NOT NULL,
  service     TEXT NOT NULL,
  endpoint    TEXT NOT NULL,
  latency_ms  DOUBLE PRECISION,
  status      INT
);
SELECT create_hypertable('metrics', 'time');

CREATE MATERIALIZED VIEW metrics_hourly
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 hour', time) AS bucket,
       service, endpoint,
       avg(latency_ms) AS avg_latency,
       percentile_cont(0.99) WITHIN GROUP (ORDER BY latency_ms) AS p99,
       count(*) AS requests
FROM metrics
GROUP BY bucket, service, endpoint;

# Prometheus recording rule + alert
groups:
  - name: api_slo
    rules:
      - record: job:http_request_duration_p99:5m
        expr: histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[5m])) by (le, job))
      - alert: HighLatencyP99
        expr: job:http_request_duration_p99:5m > 0.5
        for: 5m
        labels:
          severity: page

Inverted index

Documents are tokenized (analyzed) into terms. Each term maps to a postings list of (doc_id, positions, scores). Query "quick brown fox" intersects postings lists; BM25 ranks by term frequency and document length normalization.

flowchart LR
  d1["Doc 1: quick brown fox"]
  d2["Doc 2: slow red fox"]
  d3["Doc 3: quick rabbit"]
  idx["Inverted index"]
  t1["fox → doc1, doc2"]
  t2["quick → doc1, doc3"]
  t3["brown → doc1"]
  d1 --> idx
  d2 --> idx
  d3 --> idx
  idx --> t1
  idx --> t2
  idx --> t3

Core concepts

• Index — logical namespace (like database).
• Shard — Lucene index instance; primary + replicas for HA and read scale.
• Analyzers — tokenizer + filters (lowercase, stemming, stop words); language-specific.
• Mappings — field types (text analyzed vs keyword exact); wrong mapping requires reindex.
• Aggregations — facets, histograms, nested metrics without hitting primary DB.

PUT /products
{
  "mappings": {
    "properties": {
      "title": { "type": "text", "analyzer": "english" },
      "sku": { "type": "keyword" },
      "category": { "type": "keyword" },
      "price": { "type": "float" },
      "tags": { "type": "keyword" },
      "created_at": { "type": "date" }
    }
  }
}

GET /products/_search
{
  "query": {
    "bool": {
      "must": [{ "match": { "title": "wireless keyboard" } }],
      "filter": [{ "term": { "category": "electronics" } }]
    }
  },
  "aggs": { "by_tag": { "terms": { "field": "tags", "size": 10 } } }
}

ELK sync via Kafka (CDC pattern)

Primary DB commits → Debezium captures binlog/WAL → Kafka topic → Elasticsearch sink connector indexes documents. Near-real-time search (1–5s lag) without dual-write race conditions.

flowchart LR
  pg[(PostgreSQL)]
  debezium[Debezium CDC]
  kafka[[Kafka topic:\nproducts.changes]]
  connect[ES Sink Connector]
  es[(Elasticsearch)]
  app[Search API]
  pg -->|WAL| debezium
  debezium --> kafka
  kafka --> connect
  connect -->|bulk index| es
  app -->|query| es

Property graph model

• Nodes — entities (User, Account, Device) with properties.
• Edges — typed relationships (FOLLOWS, PAID, LOGGED_IN_FROM) with properties and direction.
• Traversal — follow pointers index-free per hop; cost ~O(edges traversed) not O(table size).

Neo4j vs Amazon Neptune

// Friends-of-friends recommendation (depth 2)
MATCH (me:User {id: $userId})-[:FOLLOWS]->(friend)-[:FOLLOWS]->(fof)
WHERE NOT (me)-[:FOLLOWS]->(fof) AND me <> fof
RETURN fof.id, count(*) AS mutualFriends
ORDER BY mutualFriends DESC
LIMIT 20;

// Fraud: shared device within 2 hops of flagged account
MATCH (flagged:Account {risk: 'high'})-[:USED_DEVICE|TRANSFERRED*1..2]-(suspect:Account)
WHERE suspect.risk IS NULL
RETURN suspect.id, count(DISTINCT flagged) AS proximityScore;

Use cases

• Social graphs — follow suggestions, feed ranking inputs (often hybrid with cache).
• Fraud / AML — connected components, cycle detection in payment graphs.
• Identity / IAM — permission inheritance, group nesting (Google Zanzibar-inspired).
• Network / IT ops — dependency mapping, blast radius analysis.
• Knowledge graphs — RAG entity linking, semantic search enrichment.

Single-leader (primary/replica)

All writes go to the leader; replicas apply the replication log (PostgreSQL WAL, MySQL binlog). Reads can hit replicas with lag. Failover promotes a replica — risk of split-brain without fencing (STONITH, etcd lease).

flowchart TB
  w1[Writer clients]
  r1[Reader clients]
  leader[(Leader DB)]
  rep1[(Replica 1)]
  rep2[(Replica 2)]
  w1 -->|writes| leader
  r1 -->|reads| rep1
  r1 -->|strong read| leader
  leader -->|async/sync replication| rep1
  leader --> rep2

• Synchronous replica — write ACK after replica confirms; survives leader loss without data loss; higher latency.
• Asynchronous replica — leader ACK before replica; faster; failover may lose last seconds.
• Semi-sync — at least one replica ACK (MySQL, PostgreSQL synchronous_standby_names).

Multi-leader (active/active)

Multiple nodes accept writes; changes replicate bidirectionally. Enables multi-region write locality (user writes to nearest DC). Write conflicts inevitable — resolve via LWW (last-write-wins), version vectors, or application merge.

flowchart LR
  dc1["DC US-East\nLeader A"]
  dc2["DC EU-West\nLeader B"]
  dc3["DC APAC\nLeader C"]
  dc1 <-->|async repl| dc2
  dc2 <-->|async repl| dc3
  dc1 <-->|async repl| dc3
  u1[US users] --> dc1
  u2[EU users] --> dc2

Leaderless (Dynamo-style quorum)

No leader — client writes to W replicas, reads from R replicas. Quorum condition R + W > N (N = replication factor) guarantees overlap: read sees at least one node with latest write (with version reconciliation).

sequenceDiagram
  participant C as Client
  participant N1 as Node 1
  participant N2 as Node 2
  participant N3 as Node 3
  Note over N1,N3: N=3, W=2, R=2
  C->>N1: write v2
  C->>N2: write v2
  N1-->>C: ACK
  N2-->>C: ACK
  Note over C: Write quorum satisfied
  C->>N2: read
  C->>N3: read
  N2-->>C: v2
  N3-->>C: v1 stale
  Note over C: Return v2 (latest timestamp wins)

Topology comparison