Partitioning

Backend Fundamentals · System Design · 9 min read

1. What problem does this solve?

Motivation
When a backend grows, a single database instance or a single table can become a bottleneck. Queries that scan large tables, write bursts, or high concurrency can overwhelm CPU, memory, or I/O, leading to slow responses or outages.

What problem exists

• Scalability limits: A single node can only handle so many reads/writes per second.
• Hot spots: Certain keys or ranges receive disproportionate traffic, causing contention.
• Maintenance pain: Large tables are hard to backup, restore, or migrate.

Why the problem matters

• Users experience latency spikes or timeouts.
• Operational costs rise as you need bigger hardware or more replicas.
• Data consistency and availability can suffer if the system is not partitioned correctly.

What happens if this concept is not used

• A monolithic table may become a single point of failure.
• Scaling horizontally becomes impossible; you’re stuck with vertical scaling.
• Long-running queries can lock the entire table, blocking other operations.

Where this concept appears in real backend systems

• Sharding a user table across multiple PostgreSQL instances.
• Partitioning a Kafka topic by key to parallelize consumer processing.
• Using Redis Cluster to distribute keyspace across nodes.
• Splitting a log table by month to keep query windows small.

2. Mental model

Analogy
Think of a library that stores all books in one massive shelf. If you want a specific book, you have to scan the whole shelf. If many people try to borrow the same book at once, the shelf becomes a bottleneck. Partitioning is like dividing the library into sections (fiction, non-fiction, science) and putting each section on a different shelf. Now you can locate books faster and multiple people can borrow books from different sections simultaneously.

Precise engineering model
Partitioning (also called sharding or splitting) divides a dataset into smaller, independent pieces called partitions or shards. Each partition is stored on a separate physical or logical resource (database instance, disk, node). The system routes a request to the correct partition based on a partition key (e.g., user ID, timestamp, hash). The goal is to keep each partition small enough to be fast, while the overall system can scale horizontally.

3. Deep explanation

Internals

• Partition key: deterministic value that maps a record to a partition (e.g., user_id % 10).
• Routing layer: middleware or client library that decides which node to query.
• Metadata store: keeps mapping of keys → nodes; may be embedded (e.g., in PostgreSQL’s pg_shard extension) or external (e.g., Consul, ZooKeeper).

Important terminology

Trade‑offs

Failure modes

• Routing failure: If the routing layer is down, queries cannot reach shards.
• Data inconsistency: If a write is sent to the wrong shard, data loss or duplication occurs.
• Hot spot: A single shard becomes overloaded, causing latency spikes.
• Rebalancing lag: During rebalancing, some reads may hit stale data.

Limitations

• Cross‑partition joins are expensive or impossible in many systems.
• Atomicity across shards is hard; you need distributed transactions or eventual consistency.
• Operational overhead: More nodes, more backups, more monitoring.

Where it fits in a backend architecture

• Data layer: Partitioned databases or key‑value stores.
• Service layer: Microservices that know how to route to the correct shard.
• Cache layer: Redis Cluster or Memcached sharding.
• Message queue: Kafka topic partitioning for parallel consumer groups.

Interaction with related concepts

• Replication: Each shard can be replicated for HA; replication + partitioning gives both scalability and availability.
• Caching: Cache keys often mirror partition keys to hit the same shard.
• CQRS: Read models can be partitioned separately from write models.
• Event sourcing: Event streams can be partitioned by aggregate ID.

4. How it works

1. Define a partition key (e.g., userId).
2. Choose a partitioning strategy (hash or range).
3. Set up metadata mapping keys to nodes.
4. Client or middleware receives a request, extracts the key, looks up the node.
5. Route the request to the appropriate node.
6. Node processes the query/write locally.
7. Return the result to the client.

sequenceDiagram
    participant Client
    participant Router
    participant Node1
    participant Node2
    Client->>Router: GET /users/12345
    Router->>Router: extract userId=12345
    Router->>Router: hash(12345) -> node=Node1
    Router->>Node1: SELECT * FROM users WHERE id=12345
    Node1-->>Router: {id:12345, name:"Alice"}
    Router-->>Client: {id:12345, name:"Alice"}

5. Practical implementation

Below is a minimal Node.js example that demonstrates hash‑based sharding for a PostgreSQL cluster.
We’ll use pg for PostgreSQL connections and a simple in‑memory routing table.

// shardRouter.ts
import { Pool } from 'pg';

type Shard = {
  id: string;
  pool: Pool;
};

const shards: Shard[] = [
  {
    id: 'shard-0',
    pool: new Pool({
      connectionString: process.env.SHARD_0_URL,
    }),
  },
  {
    id: 'shard-1',
    pool: new Pool({
      connectionString: process.env.SHARD_1_URL,
    }),
  },
];

// Simple hash function: userId % number_of_shards
function getShard(userId: number): Shard {
  const index = userId % shards.length;
  return shards[index];
}

export async function getUser(userId: number) {
  const shard = getShard(userId);
  const res = await shard.pool.query(
    'SELECT id, name, email FROM users WHERE id = $1',
    [userId]
  );
  return res.rows[0];
}

export async function createUser(user: { id: number; name: string; email: string }) {
  const shard = getShard(user.id);
  await shard.pool.query(
    'INSERT INTO users (id, name, email) VALUES ($1, $2, $3)',
    [user.id, user.name, user.email]
  );
}

Key points

• getShard uses a deterministic hash (userId % shards.length) to pick a shard.
• Each shard has its own Pool instance; connections are isolated.
• All CRUD functions route to the correct shard automatically.

Running the example

1. Spin up two PostgreSQL containers (or use two databases) and expose them on SHARD_0_URL and SHARD_1_URL.
2. Create the users table on both shards:

CREATE TABLE users (
  id BIGINT PRIMARY KEY,
  name TEXT NOT NULL,
  email TEXT NOT NULL
);

3. Run the Node.js script and call createUser / getUser.

6. Production considerations

7. Common mistakes

8. Senior engineer perspective

What changes at scale?

• Routing becomes a critical service: It must be highly available and low‑latency.
• Metadata management: Use a distributed config store (Consul, etcd) to keep shard maps.
• Rebalancing strategy: Implement incremental data migration to avoid downtime.
• Monitoring: Need fine‑grained metrics per shard, alerting on skew.
• Cost: More shards mean more infrastructure; balance cost vs. performance.

Design decisions

• Hash vs. range: For write‑heavy workloads with unpredictable keys, hash is safer. For time‑series data, range (by timestamp) is better.
• Consistent hashing: Minimizes data movement but adds complexity; use libraries like hashring.
• Replication factor: 2 or 3 replicas per shard for HA; consider synchronous vs. asynchronous replication.
• Cross‑shard transactions: Use Saga pattern or two‑phase commit if needed; otherwise, design for eventual consistency.
• Data locality: Keep related data in the same shard to avoid cross‑shard joins.

Bottlenecks & failure scenarios

• Routing layer: Single point of failure; mitigate with clustering.
• Shard hot spots: Mitigate by adding more shards or rebalancing.
• Network partitions: Use quorum reads/writes or read replicas.

Monitoring & cost

• Metrics: Query latency, shard CPU, disk I/O, replication lag.
• Cost: Each shard adds a DB instance; consider serverless or autoscaling options.

When NOT to use

• Small datasets: A single database is simpler.
• Strict ACID across all data: Partitioning complicates transactions.
• Simple read‑heavy workloads: A read replica cluster may suffice.

9. Interview questions

10. Hands‑on task

Task: Implement a simple key‑value store with hash‑based sharding in Node.js.

1. Spin up two Redis instances (or two PostgreSQL tables).
2. Create a shardRouter.ts that routes GET/SET requests to the correct instance based on the key hash.
3. Expose an HTTP API (/kv/:key) that uses the router.
4. Measure latency for keys that map to the same shard vs. different shards.
5. Add a third shard and observe how routing changes.

Deliverable: A GitHub repo with the code, a README explaining how to run, and a short report on latency observations.

11. Quick revision

• Partitioning splits data into smaller, independent units (shards).
• Hash partitioning → uniform distribution, bad for range queries.
• Range partitioning → efficient ranges, risk of hot spots.
• Consistent hashing minimizes data movement during scaling.
• Routing layer must be highly available and low‑latency.
• Replication per shard gives HA; cross‑shard transactions are hard.
• Monitor shard metrics, routing latency, and replication lag.
• Avoid partitioning small tables; use it when data or traffic grows.
• Rebalancing is expensive for range partitions; hash partitions are cheaper.
• Common pitfalls: poor hash, hard‑coded shard count, ignoring replication.

12. References

• PostgreSQL Sharding: https://www.postgresql.org/docs/current/pg_shard.html
• Redis Cluster: https://redis.io/docs/manual/scaling/cluster/
• Kafka Partitioning: https://kafka.apache.org/documentation/#partitions
• Consistent Hashing: https://en.wikipedia.org/wiki/Consistent_hashing
• Node.js pg library: https://node-postgres.com/
• Prometheus Node Exporter: https://prometheus.io/docs/instrumenting/node_exporter/
• Distributed Transactions (Saga): https://martinfowler.com/eaaDev/Saga.html