🏗️ Section 17 · Question #10

What is sharding

A Shard Key is an entity attribute (e.g., user_id, tenant_id, or country_code) used by the database driver or routing coordinator to determine which physical shard server owns a...

🏗️ Microservices 📄 Source Markdown

🟢 Junior Level

Database Sharding (Horizontal Partitioning across physical nodes) is an architectural data architecture technique where a single logical database is partitioned across multiple independent physical database servers (shards), with each server hosting a unique, disjoint subset of the total rows.

                          [SHARDED DATABASE CLUSTER]
                                       │
            ┌──────────────────────────┼──────────────────────────┐
            ▼                          ▼                          ▼
        [Shard 1]                  [Shard 2]                  [Shard 3]
       Users A – H                Users I – P                Users Q – Z
   (DB Node 1: 30M rows)      (DB Node 2: 30M rows)      (DB Node 3: 30M rows)

Read Replication vs Sharding

Dimension Read Replication Database Sharding
Data Distribution Every replica server holds 100% identical copy of the entire database. Each shard server holds only a distinct slice (e.g., 20%) of the total data.
What It Scales Scales Read Throughput (SELECT queries) across read-replicas. Scales BOTH Write Throughput (INSERT/UPDATE) and Total Storage Capacity.
Physical Limits Bound by the disk capacity and write I/O limits of the primary writer node. Scales horizontally across $N$ servers, breaking single-node hardware limits.

[!NOTE] Multi-Volume Encyclopedia Analogy: If an entire encyclopedia were bound into a single 10,000-page book, only one person could write in it at a time, and it would be too heavy to lift. By dividing it into 26 distinct volumes (A, B, C… Z), 26 people can read and write simultaneously across different volumes without blocking each other.

🟡 Middle Level

1. Shard Key Selection & Routing Strategies

A Shard Key is an entity attribute (e.g., user_id, tenant_id, or country_code) used by the database driver or routing coordinator to determine which physical shard server owns a specific row:

Strategy Routing Formula Advantages Disadvantages & Hotspot Hazards
Range-Based ID 1..1M ➔ Shard 1
ID 1M..2M ➔ Shard 2		 Extremely efficient for range queries (BETWEEN date1 AND date2). 🔴 Severe Hotspotting: All new auto-increment writes hammer the newest shard server!
Hash-Based (Standard) shard_id = hash(user_id) % N 🟢 Uniform Write Distribution: Spreads rows evenly across all shards. Range queries must be scattered to every single shard in the cluster.
Directory-Based Central lookup table:
Tenant_ABC ➔ Shard 4		 Highly flexible: High-value enterprise VIP tenants can be placed on dedicated isolated hardware. The central routing directory becomes a Single Point of Failure (SPOF) and an I/O bottleneck.

2. Java Dynamic Routing Implementation (AbstractRoutingDataSource)

In Spring Boot applications, client-side sharding can be implemented using Spring’s dynamic data source routing:

public class ShardedRoutingDataSource extends AbstractRoutingDataSource {

    @Override
    protected Object determineCurrentLookupKey() {
        // Resolve shard key from ThreadLocal security or tenant context:
        Long customerId = ShardContextHolder.getCustomerId();
        if (customerId == null) {
            return "shard_default";
        }

        // Uniform hash-based distribution across 4 physical shard databases:
        int shardIndex = Math.abs(customerId.hashCode() % 4);
        return "shard_" + shardIndex;
    }
}

3. The 3 Architectural Penalties of Sharding

  1. Scatter-Gather Queries: If a query does not supply the Shard Key (e.g., SELECT * FROM orders WHERE status = 'PENDING'), the coordinator must broadcast the query to all $N$ shards in parallel (Scatter), and then wait, merge, sort, and paginate the result sets in coordinator RAM (Gather).
  2. Loss of Foreign Key Referential Integrity: Relational databases cannot enforce FOREIGN KEY constraints across independent network nodes. An orders table on Shard 1 cannot establish an SQL foreign key against a users table on Shard 2. Referential integrity must be manually enforced in application code.
  3. Cross-Shard Distributed Transactions: Atomic updates modifying records residing on two different shards require heavy Two-Phase Commit (2PC / XA) protocols or asynchronous Saga workflows, increasing latency by $10\times\text{–}50\times$.

🔴 Senior Level

1. Distributed Primary Key Generation (ID Collisions)

In a sharded architecture, traditional database AUTO_INCREMENT or PostgreSQL SERIAL sequences cannot be used; multiple independent shards running concurrently would inevitably generate conflicting duplicate IDs (e.g., both Shard 1 and Shard 2 inserting id = 42).

Enterprise Distributed ID Architectures:

  1. Twitter Snowflake (64-bit ID): 
    [1 bit: sign] [41 bits: timestamp (ms)] [10 bits: datacenter/shard ID] [12 bits: local sequence]

    Benefits: Produces 64-bit BIGINT IDs that are globally unique, roughly time-ordered, and fit natively into standard B-Tree indexes without causing page fragmentation.

  2. UUIDv7: The modern RFC 9562 standard featuring a 48-bit Unix timestamp in its most significant bits. Unlike random UUIDv4, UUIDv7 preserves sequential index ordering and prevents index bloat.
  3. Centralized Sequence Block Allocation (Hi/Lo): Microservices fetch contiguous blocks of 1,000 IDs from a centralized Redis atomic counter (INCRBY 1000) and allocate them locally in JVM memory.

2. Consistent Hashing & Virtual Nodes (The Re-Sharding Crisis)

With naive modulo hashing (hash(key) % N), expanding a cluster from $N=3$ to $N=4$ shards causes the divisor to change for nearly all keys, resulting in $75\%$ of all historical database keys needing to be moved to different servers. This requires massive downtime and network saturation.

Consistent Hashing eliminates this catastrophe:

                  Consistent Hash Ring [0 ... 2^32 - 1]
                                 [0]
                         .----------------.
                     .'     Node A (V1)     '.
                   .'                          '.
             Node C (V2)                     Node B (V1)
                   '.                          .'
                     '.     Node A (V2)      .'
                         '----------------'
  • Keys and shard nodes are mapped onto a circular numeric space ($0 \dots 2^{32}-1$).
  • A key is stored on the first active shard node encountered clockwise along the ring.
  • When Shard $N+1$ is added, it only acquires keys from its immediate clockwise neighbor, migrating only $\frac{1}{N+1}$ of the total database without full re-indexing.
  • Virtual Nodes (vnodes): Each physical server is assigned 50–200 virtual points across the ring (e.g., NodeA-1, NodeA-2), ensuring mathematically uniform distribution of data volume and eliminating hot arcs on the ring.

3. Zero-Downtime Live Resharding

Splitting an overloaded shard on a live production database:

  1. Provisioning: Deploy and initialize the target shard instances.
  2. Dual-Write: Configure the application routing layer to write new mutations (INSERT/UPDATE/DELETE) to both old and new shards simultaneously.
  3. Backfill: A background migration worker copies historical snapshots from the old shard to the new shard.
  4. Catch-Up & CDC Verification: Tail the database transaction log via Change Data Capture (Debezium/Kafka) and perform row checksum validations to achieve consistency.
  5. Read Switch: Route read queries (SELECT) to the new shard.
  6. Stop Dual-Write: Decommission duplicate writes to the old shard and reclaim disk space.

4 Tricky Questions

1. How do you resolve the “Hotspot / Celebrity Problem” when a single tenant/user accounts for 80% of traffic on their shard?

Answer: If sharding is performed on user_id, a celebrity account with 50 million followers will overwhelm its assigned shard with read/write I/O, leaving other shards underutilized. Solutions:

  1. Salted Shard Key: For detected high-volume entities, append a randomized salt: key = user_id + "_" + random(1..10). The celebrity’s data is partitioned across 10 separate physical shards. Read queries perform a fast parallel scatter-gather across those 10 shards.
  2. Dedicated VIP Sharding: Use Directory-Based sharding to isolate high-throughput enterprise tenants onto dedicated physical database clusters.
  3. Aggressive Read Caching: Offload celebrity read traffic entirely to an in-memory Redis/Memcached cluster.

2. What is the catastrophic danger of deep pagination (ORDER BY created_at LIMIT 10 OFFSET 100000) in a sharded database?

Answer: When querying without a Shard Key, the query coordinator must scatter the request to all $N$ shards. To return 10 records at offset 100,000, each shard must independently evaluate its local data and return its top $100,010$ matching records. With 10 shards, the coordinator must transfer over $1,000,000$ rows over the network, buffer them in JVM memory, sort the merged collection, discard the first 100,000 rows, and return 10. This leads to massive network saturation, coordinator OutOfMemoryError, and multi-second latency spikes. Remedy: Prohibit OFFSET pagination on sharded datasets and enforce Keyset Pagination (Seek Method):

WHERE (created_at, id) < (:last_seen_created_at, :last_seen_id) 
ORDER BY created_at DESC, id DESC LIMIT 10;

3. Why are global UNIQUE constraints (on non-shard-key fields like email) extremely difficult to enforce in a sharded database?

Answer: Relational databases enforce UNIQUE constraints via a local B-Tree index stored on the local storage engine. If a table is sharded by user_id, Shard 1 has no visibility into whether an email value already exists on Shard 2. Enforcing global uniqueness at the database level would require an expensive distributed lock across all shards prior to every INSERT. Remedies:

  • Maintain a separate global index table (e.g., email_to_user_id) sharded by hash(email). The application inserts into this index table first to claim the unique email before creating the user on the primary user shard.
  • Use an in-memory distributed lock (Redis/Redlock) or dedicated Identity Provider (Keycloak).

4. What is the fundamental trade-off between Application-Level Sharding and Database-Level Distributed SQL (NewSQL)?

Answer:

  • Application-Level Sharding (Apache ShardingSphere, Citus, Custom Routers): Runs on top of standard, inexpensive commodity PostgreSQL/MySQL instances. The application or an intermediate proxy manages routing logic. Trade-off: High application complexity, manual resharding operations, and complex cross-shard transaction support.
  • Distributed NewSQL (CockroachDB, YugabyteDB, Google Cloud Spanner): The database handles sharding (ranges/tablets) natively using Raft/Paxos consensus, providing transparent multi-node ACID transactions and automated self-healing rebalancing. Trade-off: Requires high-performance low-latency network interconnects between nodes, higher CPU/memory footprints, and higher operational licensing costs.

🎯 Interview Cheat Sheet

Must Know

  • Definition: Horizontal partitioning across distinct physical database servers where each server owns a unique slice of data rows.
  • Core Purpose: Scales write throughput and total data volume beyond the hardware limits of a single machine.
  • Shard Key: The routing attribute (e.g., user_id); must provide high cardinality and uniform write distribution.
  • Routing Strategies: Hash-based (uniform distribution), Range-based (fast range queries, but prone to write hotspots), Directory-based (flexible VIP tenant placement).
  • Consistent Hashing: Uses a logical hash ring ($0 \dots 2^{32}-1$) with virtual nodes to minimize data migration when expanding shard nodes.
  • ID Strategy: Use 64-bit Twitter Snowflake or UUIDv7; database AUTO_INCREMENT produces collisions across shards.

The 3 Big Sharding Penalties

  1. Scatter-Gather Latency: Queries lacking a shard key must query all shards and merge in memory.
  2. Broken Foreign Keys: Relational engines cannot enforce cross-node foreign key integrity.
  3. Cross-Shard Transactions: Require distributed 2PC/XA or Saga patterns.

Common Follow-up Questions

  • “When should you shard?” $\to$ Only as a last resort when vertical scaling (scale-up), read replicas, caching, and table partitioning on a single instance are exhausted (typically $>1\text{–}2\text{ TB}$ or $>50\text{k write TPS}$).
  • “What is a virtual node in consistent hashing?” $\to$ Mapping a single physical machine to multiple positions on the hash ring to ensure uniform load distribution.
  • “Why can’t you use SQL OFFSET pagination?” $\to$ Coordinator must pull OFFSET + LIMIT rows from every single shard, causing OOM and network congestion.

Red Flags (DO NOT Say)

  • ❌ “We should shard our database from day 1 on a brand new project.” (Premature optimization that introduces immense architectural overhead).
  • ❌ “Sharding and Read Replication are identical concepts.” (Replication clones 100% of data for reads; Sharding partitions distinct subsets of data for writes and storage).
  • ❌ “Auto-increment primary keys work fine in sharded databases.” (Auto-increment generates duplicate IDs across different shard nodes).