Bird
Raised Fist0

Database Sharding Strategies - Range, Hash, Directory-Based

HLDmediumInterview freq: common
Choose your preparation mode4 modes available
ArchitectureTraceChallenge
🏗️
Database Sharding Strategies - Range, Hash, Directory-Based
Design and evaluate database sharding strategies for scalable transactional systems
mediumDatabases - SQL and TransactionsFreq: common
⚙️
Functional Requirements
  • Support horizontal scaling of a relational database to handle billions of users and high QPS.
  • Implement sharding strategies: range-based, hash-based, and directory-based partitioning.
  • Ensure strong transactional consistency (ACID) within shards.
  • Support efficient query routing to the correct shard based on shard key.
  • Allow online resharding or shard rebalancing with minimal downtime.
  • Provide mechanisms for shard metadata management and discovery.
📊
Non-Functional Requirements
  • Handle 2 billion daily active users (DAU) generating 40 messages per user per day.
  • Achieve 99.99% availability for read and write operations.
  • Maintain 100ms P99 latency for transactional queries.
  • Support data retention of 30 days with replication factor of 3 for durability.
  • Scale to peak write QPS of approximately 925K writes/sec.
❓
Clarifying Questions to Ask
  • What is the primary shard key or keys used for partitioning?
  • Are cross-shard transactions required or only single-shard transactions?
  • What is the expected read-to-write ratio for the workload?
  • Is the workload mostly point lookups, range scans, or analytical queries?
  • How frequently do shards need to be rebalanced or resharded?
  • What is the tolerance for stale reads or eventual consistency?
Out of Scope
✗ Designing the full database engine internals (e.g., storage engine, query optimizer).
✗ Designing cross-shard distributed transactions or global consistency protocols.
✗ Designing the application layer or client SDKs for sharding.
✗ Backup and disaster recovery strategies beyond replication.
📌
Assumptions
  • 2 billion DAU
  • 40 messages per user per day
  • 100 bytes per message average size
  • Replication factor of 3
⚡
QPS Calculation
write qps2B × 40 / 86400 ≈ 925,926 writes/sec
peak write qpsAssuming 3× peak factor: 925,926 × 3 ≈ 2.78M writes/sec
read qpsAssuming 10 reads per write: 925,926 × 10 ≈ 9.26M reads/sec
💾
Storage Estimate
daily storage2B × 40 × 100 bytes = 8 TB/day uncompressed
retention 30 days8 TB × 30 = 240 TB
with replication240 TB × 3 = 720 TB total storage
🖥️
Server Estimate
write db nodesAssuming each node handles 10K writes/sec, need 2.78M / 10K = 278 nodes
read db nodesAssuming each node handles 50K reads/sec, need 9.26M / 50K = 185 nodes
🏗️
Architecture Diagram
Architecture Diagram
                +----------------------+
                |   Client Applications |
                +----------+-----------+
                           |
                           v
                +----------------------+
                |  Query Router / Proxy |
                +----------+-----------+
                           |
          +----------------+----------------+----------------+
          |                |                |                |
          v                v                v                v
   +------------+   +------------+   +------------+   +------------+
   | Shard 1 DB |   | Shard 2 DB |   | Shard 3 DB |   | Shard N DB |
   +------------+   +------------+   +------------+   +------------+
          |                |                |                |
          v                v                v                v
   +------------+   +------------+   +------------+   +------------+
   |Replica Set |   |Replica Set |   |Replica Set |   |Replica Set |
   +------------+   +------------+   +------------+   +------------+
                           |
                           v
                +----------------------+
                |  Shard Metadata Store |
                +----------------------+
🧩
Components
Query Router / Proxy
Routes client queries to the correct shard based on shard key
Why thisUses Envoy proxy with custom sharding plugin for low-latency routing and dynamic shard metadata refresh
Why notDirect client routing rejected due to complexity and inability to centralize shard metadata updates
Shard Metadata Store
Stores shard mapping metadata for directory-based sharding and shard discovery
Why thisUses Apache Zookeeper for consistent, highly available metadata storage with watch notifications
Why notRejected etcd due to higher latency under heavy watch load at 10M+ clients
Shard DB Nodes
Hosts partitioned data shards with transactional support
Why thisUses Google Spanner for global strong consistency and horizontal scalability with built-in sharding
Why notRejected PostgreSQL due to manual sharding complexity and write throughput limits at 925K writes/sec
Replica Sets
Provide read scalability and high availability for each shard
Why thisUses Spanner’s multi-region replicas for synchronous replication and 99.99% availability
Why notAsynchronous replicas rejected due to unacceptable stale reads for transactional workloads
Shard Rebalancer Service
Manages shard splits, merges, and resharding with minimal downtime
Why thisCustom service leveraging consistent hashing and metadata updates to redistribute data incrementally
Why notRejected offline resharding due to unacceptable downtime and data unavailability
🔄
Data Flow
1
Client Applications → Query Router / Proxy
Client sends query with shard key
Router uses shard key to determine target shard
2
Query Router / Proxy → Shard Metadata Store
Fetch shard mapping if not cached
Ensures routing correctness with up-to-date shard info
3
Query Router / Proxy → Shard DB Nodes
Forward query to correct shard node
Single-shard transactional queries executed here
4
Shard DB Nodes → Replica Sets
Synchronously replicate data for durability
Maintains strong consistency and availability
5
Shard Rebalancer Service → Shard Metadata Store
Update shard mappings after resharding
Clients and router get updated shard info dynamically

Practice

(1/5)
1. In the event sourcing system trace, at Step 4 the Delivery Worker reads an event from Kafka and attempts to update the read model. What happens if the read model update fails due to a transient database error?
easy
A. The Delivery Worker sends the event back to Kafka for reprocessing
B. The Delivery Worker discards the event and moves to the next one
C. The Delivery Worker commits the offset immediately and logs the failure
D. The Delivery Worker retries the update until success before committing the offset

Solution

  1. Step 1: Trace Delivery Worker decision at Step 4 -- read model update

    The Delivery Worker must ensure the read model is updated before committing the Kafka offset to avoid data loss.

  2. Step 2: Understand failure handling

    On transient DB errors, the worker retries to guarantee eventual consistency; discarding or committing offset prematurely risks data loss.

  3. Final Answer:

    Option D -> Option D

  4. Quick Check:

    Retry read model update before offset commit -> ensures consistency [check]
Hint: Commit Kafka offset only after successful read model update [check]
Common Mistakes:
2. Without the bulkhead pattern implemented, what exactly happens when one microservice's database connection pool saturates under load?
medium
A. The entire system queues requests indefinitely, causing high latency everywhere
B. Requests to the saturated service block and cause thread starvation, cascading failures to other services
C. The service automatically rejects requests, preventing impact on other services
D. Load balancers detect saturation and reroute traffic to healthy instances instantly

Solution

  1. Step 1: Understand absence of bulkhead isolation

    Without bulkheads, saturation causes blocking (B) and thread starvation, but automatic rejection (C) is a bulkhead feature. Queuing indefinitely (A) is unlikely without bulkheads. Load balancers (D) don't detect internal resource saturation.

  2. Step 2: Trace failure impact

    Without bulkheads, blocking threads cause cascading failures; with bulkheads, requests are rejected to contain blast radius.

  3. Final Answer:

    Option C -> Option C

  4. Quick Check:

    Without bulkheads, blocking causes cascading failures [check]
Hint: No bulkheads -> blocking threads cascade failures [check]
Common Mistakes:
3. For analytics on URL redirects, you must track click counts per short code with minimal latency and high write throughput. Which storage solution is best suited for this requirement?
medium
A. Use an in-memory data store like Redis with atomic increment operations and periodic batch writes to persistent storage
B. Use a relational database with row-level locking for click count updates
C. Use a distributed log system like Kafka to store every click event and process counts offline
D. Use a NoSQL document store with eventual consistency for click count updates

Solution

  1. Step 1: Analyze latency and throughput needs

    High write throughput and low latency require fast atomic increments without locking overhead.

  2. Step 2: Evaluate storage options

    Relational DB (A) suffers from locking; Kafka (C) adds latency due to offline processing; NoSQL (D) eventual consistency may cause inaccurate counts. Redis (B) supports atomic increments and fast writes with batch persistence.

  3. Final Answer:

    Option A -> Option A

  4. Quick Check:

    Redis atomic increments enable low-latency, high-throughput analytics [check]
Hint: Redis atomic increments for fast, accurate click counts [check]
Common Mistakes:
4. A company must expose internal web services securely to external clients while hiding internal server details and enabling SSL termination. Which proxy architecture is best suited for this?
medium
A. Use a reverse proxy to terminate SSL and hide internal server details
B. Use a transparent proxy to intercept and forward requests without modification
C. Use a forward proxy to route external client requests to internal servers
D. Use a DNS proxy to resolve internal server names for external clients

Solution

  1. Step 1: Identify proxy role for exposing internal services

    A reverse proxy sits in front of internal servers, hides their details, and can terminate SSL for external clients.

  2. Step 2: Eliminate other options

    Forward proxies act on clients, not servers; transparent proxies don't modify requests; DNS proxies only resolve names, not handle SSL or hide servers.

  3. Final Answer:

    Option A -> Option A

  4. Quick Check:

    Reverse proxy hides servers and terminates SSL [check]
Hint: Reverse proxy hides servers and handles SSL termination [check]
Common Mistakes:
5. At 10x scale, which component in a cache stampede prevention system is most likely to fail first, and why?
hard
A. Cache servers, because they run out of memory storing more keys
B. Backend database, because it cannot handle the increased read load from cache misses
C. Distributed lock service, because lock contention and coordination overhead grow exponentially
D. Load balancers, because they cannot distribute requests evenly at high concurrency

Solution

  1. Step 1: Identify bottlenecks at 10x scale

    Backend and cache scale horizontally; load balancers scale well.

  2. Step 2: Analyze distributed lock service

    Lock service faces increased contention and coordination overhead, causing latency and failures.

  3. Final Answer:

    Option C -> Option C

  4. Quick Check:

    Distributed lock service bottlenecks first due to coordination overhead [check]
Hint: Lock service bottlenecks first due to exponential coordination overhead [OK]
Common Mistakes:
Recommended

Ready to prove it?

Pass the test and this pattern moves unknown strong while earning +1 Grip toward your certificate.

10 questions~5 mineasy hard