Backend / Databases / SQL / 14_sharding_partitioning.md

Sharding and Partitioning

Updated 10 interview angles 12 min read source
On this page13
  1. The distinction at a glance
  2. Partitioning (single DB)
  3. Sharding (multiple DBs)
  4. Distributed transactions — the painful part
  5. Replication vs sharding — different problems
  6. Choosing the shard key — the single most important decision
  7. Real-world examples
  8. Schema migrations across shards
  9. When to shard (and when not to)
  10. Tools and approaches
  11. Common pitfalls
  12. Common interview confusions
  13. Interview angle

Sharding and Partitioning

Both split a table into smaller pieces. Partitioning keeps the pieces within one database. Sharding splits across machines. Different problems, different trade-offs, often confused in interviews.

For when each fits, distributed transaction trade-offs, and operational reality.

The distinction at a glance

Partitioning Sharding
Scope one DB instance many DB instances
Routing DB engine application or proxy
Cross-piece queries normal SQL hard (scatter-gather)
Transactions across pieces yes (one DB) no (or 2PC; expensive)
Operational complexity low high
Use for manageable single-DB performance scale beyond one machine

Partitioning first; shard only when you have to.

Partitioning (single DB)

One logical table; multiple physical pieces. The DB engine routes queries to the relevant partition(s).

Vertical partitioning

Split columns into separate tables, joined on PK.

sql
-- "hot" frequently-read columns
CREATE TABLE users_core (
    id BIGINT PRIMARY KEY,
    email TEXT,
    name TEXT
);

-- "cold" rarely-read columns
CREATE TABLE users_profile (
    id BIGINT PRIMARY KEY REFERENCES users_core,
    bio TEXT,
    avatar BYTEA          -- could be megabytes
);

Win: hot table fits in cache; rare reads pay the JOIN cost. Useful for tables with a mix of small-frequent and large-rare columns.

Horizontal partitioning (Postgres declarative)

Split rows across child tables.

sql
CREATE TABLE events (
    id BIGINT,
    user_id BIGINT,
    created_at TIMESTAMPTZ NOT NULL,
    payload JSONB
) PARTITION BY RANGE (created_at);

CREATE TABLE events_2026_01 PARTITION OF events
    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
CREATE TABLE events_2026_02 PARTITION OF events
    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');

Partitioning strategies:

Strategy When
RANGE (col) continuous values: date, ID, numeric range
LIST (col) discrete enum-like values: region, tenant_id (small N)
HASH (col) even distribution, no semantic meaning (write fan-out)

Wins from partitioning

  • Partition pruningWHERE created_at >= '2026-02-01' only scans the Feb partition.
  • Cheap dropsDROP TABLE events_2025_12 is instant; DELETE WHERE created_at < ... would take hours.
  • Smaller indexes per partition — each partition’s index fits in cache.
  • Parallel maintenance — vacuum / analyze / reindex per partition concurrently.
  • Sequential scans on hot partition — when you’re only touching recent data.

Pitfalls

  • Each query must include the partition key in WHERE for pruning to kick in. EXPLAIN to verify.
  • UNIQUE constraints must include the partition key.
  • Foreign keys referencing a partitioned table need PG 12+.
  • Cross-partition UPDATE (changing a row’s partition key) needs PG 11+.
  • Too many partitions (>1000) cause planner overhead.

When to partition

  • Time-series data with old data dropped regularly.
  • Tables exceeding ~10 GB where most queries scope to a slice.
  • Large tables where index size matters.
  • Audit logs / events with predictable archival.

For a million-row table queried by user_id, partitioning probably won’t help. For a billion-row time-series with WHERE created_at >= '...' queries, it’s transformative.

Sharding (multiple DBs)

Data spans multiple database instances. Application or proxy chooses where to read/write.

Sharding strategies

Strategy How shard chosen Use when
Hash hash(key) % N Even distribution; no range queries needed
Range key BETWEEN A AND B → shard X Range queries common; sequential keys
Directory Lookup table: key → shard Tenants with very different sizes
Geographic Region of user → local shard Latency, data residency, compliance

Each strategy trades off: hash is simple but range queries hit every shard; range allows efficient ranges but creates hot spots; directory adds operational complexity.

Consistent hashing — solving the resharding problem

Naive hashing (hash(key) % N): adding a shard means rehashing — every key may move. Catastrophic at scale.

Consistent hashing: nodes and keys map to points on a hash ring. Each key goes to the nearest node clockwise. Adding a node only moves keys in its arc — ~K/N keys, not all of them.

text
       0
       |
   N1--+--N2
       |
       N3
       |
     (key X → walks clockwise → lands on N2)

Add N4:
       0
       |
   N1--+--N4
       |    \
       |     N2
       N3
       
   Only keys between N1 and N4 move (formerly went to N2).

Virtual nodes: each physical node owns many points on the ring. Smooths distribution and makes adding/removing nodes affect many keys proportionally.

Used by: DynamoDB, Cassandra, Riak, Memcached client libraries.

Pre-sharding

Hash to many virtual shards (e.g., 1024) up front; place virtual shards on physical nodes. Rebalancing moves virtual shards, not individual keys.

text
1024 virtual shards
4 physical nodes initially: each owns 256 virtual shards
Add a node → migrate 1024/5 ≈ 205 virtual shards to the new node

Easier to operate than pure consistent hashing. Used by Vitess, Discord, Slack.

Cross-shard queries are hard

Once data spans shards:

JOINs across shards: application-side scatter-gather:

python
# Pseudocode
def get_user_with_orders(user_id):
    user_shard = shard_for(user_id)
    user = user_shard.query("SELECT * FROM users WHERE id = %s", user_id)
    # Orders might be on different shards (if sharded by order_id)
    all_shards = list_shards()
    orders = []
    for shard in all_shards:
        orders.extend(shard.query("SELECT * FROM orders WHERE user_id = %s", user_id))
    return user, orders

Co-locate by shard key to avoid this: shard both users and orders by user_id so they’re on the same shard.

Aggregates: scatter-gather + merge in application:

python
def total_orders():
    return sum(shard.query("SELECT count(*) FROM orders") for shard in shards)

For sums, counts, averages this works. For percentiles, top-K, complex group-by, it gets painful. Use OLAP tools (BigQuery, ClickHouse) for analytics; don’t try to do them on sharded OLTP DBs.

Unique constraints: only enforced within a shard. Use globally-unique IDs:

  • UUIDs (random, easy, larger).
  • Snowflake-style 64-bit IDs (timestamp + machine + sequence).
  • A central ID-generation service.

Foreign keys: broken across shards. Replace with soft references (just store the ID; no DB-level FK).

Distributed transactions — the painful part

You write to two shards in one user action. How do you keep them atomic?

Two-Phase Commit (2PC)

text
1. Coordinator → Shards: PREPARE
2. Each shard locks rows, writes to WAL, replies VOTE_COMMIT or VOTE_ABORT
3. Coordinator: if all vote commit → COMMIT; else → ABORT
4. Each shard finalizes

Problems:

  • Slow — multiple round trips, locks held throughout.
  • Coordinator is SPOF — if it crashes after phase 1, shards hold locks indefinitely until manual intervention.
  • Increasing failure surface — N+1 things that can fail (each shard plus coordinator).
  • Operational nightmare — recovery from partial commits is painful.

Modern systems avoid 2PC for OLTP. Use it only for rare, critical operations where atomicity is strictly required.

Saga pattern — eventual consistency alternative

Series of local transactions, each with a compensating action if a later one fails.

text
1. Reserve inventory               → if fail, abort
2. Charge payment                  → if fail, release inventory
3. Create shipment                 → if fail, refund payment, release inventory
4. Send confirmation email         → if fail, log; no compensation

Each step is a local transaction on one shard. Failure triggers compensating actions to roll back earlier steps. Eventually consistent.

Pros: no 2PC overhead; scales horizontally. Cons: complex to implement correctly; compensations can themselves fail.

See Event-Driven Architecture and Sagas.

Outbox pattern — atomic DB + event publishing

sql
BEGIN;
INSERT INTO orders (...);
INSERT INTO outbox (event_type, payload) VALUES ('OrderCreated', '{...}');
COMMIT;

A separate process reads from outbox and publishes events to other shards / services. Atomic within one DB transaction; eventually consistent across shards.

Used heavily in microservices + event-driven architectures.

Replication vs sharding — different problems

Often confused. They solve different scaling axes:

Replication Sharding
Scale reads reads + writes
Each instance holds full data copy a slice of the data
Failover promote a replica shard goes down → that data unavailable
Complexity low high
Use when read-heavy, can tolerate staleness dataset doesn’t fit one machine

Most apps scale fine with read replicas for read-heavy workloads. Sharding is for when one machine can’t hold the data (multi-TB) or handle the write throughput.

Both can coexist: shard for capacity, replicate each shard for HA.

Choosing the shard key — the single most important decision

Bad shard key = hot shard = the whole point lost.

Good keys:

  • High cardinality — many distinct values so traffic spreads.
  • Even access distribution — no value gets disproportionate load.
  • Co-locates data accessed together — a user’s data on one shard, not spread across all.
  • Stable — doesn’t change, so re-sharding isn’t constant.

Common bad keys:

Bad key Why
country 50% of traffic ends up on one shard
created_at all writes go to the most recent shard (“hot tail”)
Sequential IDs same hot-tail problem
status 4 values → uneven distribution
Composite of unrelated cols unpredictable distribution

For SaaS multi-tenant: tenant_id is usually right. Each tenant’s data is co-located; queries scoped to one tenant hit one shard.

For social/consumer: user_id. Their posts, comments, likes — co-located.

Real-world examples

Instagram: shard by user_id with pre-sharding (logical → physical mapping). They started with one Postgres and outgrew it; the sharding strategy let them scale to billions of users without rewriting.

Discord: ScyllaDB (Cassandra-compatible) sharded by user_id. Migrated from MongoDB after hitting scaling limits.

Pinterest: Sharded MySQL with consistent hashing. Each shard ID is a 64-bit composite of (shard, type, local_id), letting them shard once and not worry about cross-shard references.

Slack: Vitess (MySQL sharding) sharded by workspace_id. Each workspace’s data is co-located.

Twitter (X): Manhattan + Gizzard. Shard by user_id for most things; timelines fanned out via separate services.

Common pattern: shard by the natural “tenant” boundary (user, workspace, organization). Avoids cross-shard queries for the 99% case.

Schema migrations across shards

In a single DB: ALTER TABLE ... ADD COLUMN ... runs once.

In a sharded system:

  • Run the migration on every shard.
  • Coordinate when each shard is at which version.
  • Code must handle “some shards have new column, some don’t” during the migration window.

Tools (Vitess, Gh-ost, pt-online-schema-change) help. Operationally, expect schema changes to take days in a sharded fleet vs minutes in a single DB.

For zero-downtime migrations see Zero-downtime migrations.

When to shard (and when not to)

Don’t shard until you’ve exhausted:

  1. Indexes + query optimization (EXPLAIN and EXPLAIN ANALYZE).
  2. Read replicas for read-heavy workloads.
  3. Connection pooling (Connection pooling).
  4. Vertical scaling — modern hardware is huge (128 cores, 1+ TB RAM, NVMe).
  5. Caching (Redis in front).
  6. Partitioning within one DB.
  7. NoSQL alternatives if your workload fits.

Sharding adds operational complexity that compounds: backups, failover, schema migrations, debugging, observability, monitoring — everything 10× harder.

The classic mistake: sharding “for scale” before hitting the actual ceiling. Most workloads never reach single-DB limits on modern hardware (16-core, 256 GB RAM, NVMe — easily 50k QPS read-mostly, 10k QPS write-mostly).

When you actually need to shard:

  • Data exceeds single-machine capacity (multi-TB, growing).
  • Write throughput exceeds what one machine can sustain.
  • Geographic distribution required (data residency, low latency).
  • Fault isolation — one tenant’s traffic shouldn’t affect others.

Tools and approaches

Tool What
Citus Postgres extension; turns Postgres into a distributed DB. SQL stays mostly unchanged.
Vitess MySQL sharding. Used by YouTube, Slack.
CockroachDB Distributed SQL, sharding built-in. PostgreSQL-compatible wire protocol.
YugabyteDB Distributed SQL, Postgres-compatible.
Cassandra / ScyllaDB NoSQL, native sharding via consistent hashing.
DynamoDB AWS managed; partition key drives sharding.
Spanner Google’s distributed SQL.
Application-side Your code routes queries; most flexible, most code.

For new projects needing horizontal scale: distributed SQL (CockroachDB, Yugabyte, Spanner) gives you SQL semantics without manual sharding.

For existing single-DB systems: extend with Citus, Vitess, or migrate to a sharded NoSQL.

For new microservices: shard at the service boundary (each service owns its data).

Common pitfalls

  • Sharding too early — operational cost without payoff.
  • Wrong shard key — hot shard, can’t rebalance.
  • Cross-shard transactions everywhere — defeats the point.
  • Not co-locating related data — every query becomes scatter-gather.
  • Foreign keys across shards — broken; need to redesign as soft references.
  • Single-shard hot key — celebrity tenant with 90% of traffic on one shard. Need rebalancing or directory-based sharding.
  • Resharding under load — design for it from day one (consistent hashing or pre-sharding).
  • Migration synchronization — schema drift across shards.

Common interview confusions

  • “Sharding and partitioning are the same.” — partitioning keeps the pieces in one DB; sharding spans machines.
  • “Sharding fixes performance.” — sharding adds operational complexity. It scales capacity, not raw query speed.
  • “NoSQL means sharded.” — many NoSQL DBs shard automatically (Cassandra, MongoDB), but it’s a separate concept; sharded SQL exists too.
  • “Just use Postgres until 1 TB.” — depends on workload. Some apps hit limits earlier; some go far beyond. Measure, don’t guess.

Interview angle 10

  • “Difference between partitioning and sharding?” — partitioning splits a table within one DB; sharding splits across machines. Partition queries go through the same DB engine; sharded queries are routed by application or proxy.
  • “When would you choose to shard?” — when data exceeds single-machine capacity (multi-TB and growing), write throughput exceeds one machine, geographic distribution required for latency/compliance, or fault isolation between tenants is needed. Exhaust replicas + vertical scaling + partitioning + caching first.
  • “How do you pick a shard key?” — high cardinality, even distribution, co-locate data accessed together, stable. SaaS: tenant_id; social: user_id; avoid sequential timestamps or low-cardinality fields.
  • “What’s consistent hashing?” — hash function that maps keys and nodes to points on a ring; each key goes to the nearest node clockwise. Adding/removing a node only moves ~K/N keys instead of all keys. Used by DynamoDB, Cassandra.
  • “What’s the difference between consistent hashing and pre-sharding?” — consistent hashing is dynamic (smooth migrations on every change). Pre-sharding creates many virtual shards up front, maps them to physical nodes; rebalancing moves whole virtual shards. Easier to operate; used by Vitess, Discord, Slack.
  • “How do you handle cross-shard transactions?” — three options: 2PC (slow, fragile, avoid for OLTP), sagas with compensating actions (eventually consistent, complex), or co-locate the affected data so transactions stay within one shard (preferred).
  • “What’s the difference between sharding and replication?” — replication: every node holds full data; scales reads. Sharding: each node holds a slice; scales capacity + write throughput. Often combined: shard for capacity, replicate each shard for HA.
  • “Schema migrations in a sharded system — how?” — run the migration on each shard; coordinate progress; code must tolerate mixed schema versions during migration. Tools (Vitess, Gh-ost) help. Expect days/weeks vs minutes in single-DB.
  • “What’s an outbox pattern?” — write to your DB and an outbox table in the same transaction; a separate process reads outbox and publishes events. Atomic locally; eventually consistent globally.
  • “Real-world sharded systems you know?” — Instagram (sharded Postgres by user_id), Discord (ScyllaDB by user_id), Slack/YouTube (Vitess on MySQL), Twitter (Manhattan), DynamoDB (consistent hashing).

See CAP, ACID, BASE, PACELC for the consistency trade-offs sharding introduces. For event-driven coordination see Event-Driven Architecture and Sagas.