Database Sharding Strategies: Horizontal Partitioning for Scale

At some point, your database becomes the thing every engineer on your team watches. Query times climb. Replication lag grows. The ops team talks about “bigger boxes.” You add read replicas and buy some time, but writes still funnel through one primary. You add connection pooling. You tune indexes until there is nothing left to tune. Eventually, the ceiling is the ceiling.

Sharding is how you break through it. But sharding is also how you introduce a category of problems that can occupy your team for years if you do it without a clear strategy. This guide covers the four major sharding strategies, the tradeoffs that matter in production, and a framework for deciding whether you need sharding at all.

Why Single-Database Architectures Hit a Wall

A single primary database eventually runs into fundamental constraints:

Write throughput. Every write goes to one node. No amount of hardware acceleration changes the fact that a single write path has a ceiling, whether that ceiling is disk I/O, network bandwidth, lock contention, or WAL write speed.

Working set size. When your hot dataset exceeds available RAM, cache hit rates drop and every query starts touching disk. Vertical scaling (more RAM) works until it doesn’t, and the jumps in cost become severe.

Lock contention. High-concurrency workloads on shared tables create lock pressure. This is not an indexing problem. It is a concurrency problem that only partitioning solves.

Replication lag. Read replicas help read-heavy workloads, but replica lag under heavy write loads creates stale read problems that are hard to paper over at the application layer.

When you hit two or more of these simultaneously, you are in sharding territory.

The Four Major Sharding Strategies

Hash-Based Sharding

Hash-based sharding computes a hash of the shard key and assigns the record to a shard using modulo arithmetic. The goal is even distribution regardless of data patterns.

function getShard(userId: string, shardCount: number): number {
  // FNV-1a hash for deterministic, well-distributed results
  let hash = 2166136261;
  for (let i = 0; i < userId.length; i++) {
    hash ^= userId.charCodeAt(i);
    hash = (hash * 16777619) >>> 0; // unsigned 32-bit multiply
  }
  return hash % shardCount;
}

class ShardedUserRepository {
  constructor(
    private shards: Pool[],
  ) {}

  private getPool(userId: string): Pool {
    const shardIndex = getShard(userId, this.shards.length);
    return this.shards[shardIndex];
  }

  async findUser(userId: string): Promise<User | null> {
    const pool = this.getPool(userId);
    const result = await pool.query(
      'SELECT * FROM users WHERE id = $1',
      [userId]
    );
    return result.rows[0] ?? null;
  }

  async createUser(user: User): Promise<void> {
    const pool = this.getPool(user.id);
    await pool.query(
      'INSERT INTO users (id, email, created_at) VALUES ($1, $2, $3)',
      [user.id, user.email, user.createdAt]
    );
  }
}

Property	Assessment
Distribution	Even by design; no hot shards from data skew
Range queries	Poor: rows with adjacent keys land on different shards
Resharding	Expensive: changing shard count remaps most keys
Failure isolation	Good: one shard down affects only that key range
Implementation complexity	Low

Failure modes: Naive modulo sharding is brittle during resharding. If you move from 4 to 5 shards, nearly every record needs to move. Use consistent hashing if you anticipate scaling the shard count. Also watch for hotspot formation when shard keys are not high-cardinality (e.g., sharding on a country column with 3 values and 8 shards).

Range-Based Sharding

Range-based sharding assigns contiguous key ranges to shards. This is natural for time-series data, ordered IDs, or anything where you frequently query by range.

interface ShardRange {
  shardId: number;
  minKey: string;
  maxKey: string;
  pool: Pool;
}

class RangeShardRouter {
  constructor(private ranges: ShardRange[]) {
    // Ranges must be sorted by minKey and non-overlapping
    this.ranges.sort((a, b) => a.minKey.localeCompare(b.minKey));
  }

  getPool(key: string): Pool {
    for (const range of this.ranges) {
      if (key >= range.minKey && key < range.maxKey) {
        return range.pool;
      }
    }
    throw new Error(`No shard found for key: ${key}`);
  }

  async queryRange(startKey: string, endKey: string): Promise<Row[]> {
    // Identify which shards overlap with the requested range
    const involvedShards = this.ranges.filter(
      (r) => r.maxKey > startKey && r.minKey < endKey
    );

    const queries = involvedShards.map((shard) =>
      shard.pool.query(
        'SELECT * FROM events WHERE event_id >= $1 AND event_id < $2 ORDER BY event_id',
        [
          startKey > shard.minKey ? startKey : shard.minKey,
          endKey < shard.maxKey ? endKey : shard.maxKey,
        ]
      )
    );

    const results = await Promise.all(queries);
    return results.flatMap((r) => r.rows);
  }
}

Property	Assessment
Distribution	Uneven if data is skewed toward recent keys (time-series hotspot)
Range queries	Excellent: adjacent keys co-locate on the same shard
Resharding	Moderate: split a shard range, migrate one shard’s data
Failure isolation	Good, but write hotspot risk on latest shard
Implementation complexity	Moderate

Failure modes: Time-series workloads with range sharding almost always create a “hot tail” problem. All writes land on the shard holding the current time range while older shards sit idle. Solutions include time-bucketing with rotation (pre-create shards for future time windows), or combining range sharding on time with hash sharding on a secondary key.

Directory-Based Sharding

Directory-based sharding maintains an explicit lookup table that maps keys to shards. The shard assignment is arbitrary and can be updated without rehashing or range changes.

interface ShardMapping {
  tenantId: string;
  shardId: number;
  region: string;
}

class DirectoryShardRouter {
  private cache = new Map<string, ShardMapping>();

  constructor(
    private directoryPool: Pool,
    private shards: Map<number, Pool>
  ) {}

  async getPool(tenantId: string): Promise<Pool> {
    // Check in-memory cache first
    let mapping = this.cache.get(tenantId);

    if (!mapping) {
      const result = await this.directoryPool.query<ShardMapping>(
        'SELECT tenant_id, shard_id, region FROM shard_directory WHERE tenant_id = $1',
        [tenantId]
      );

      if (result.rows.length === 0) {
        throw new Error(`No shard mapping for tenant: ${tenantId}`);
      }

      mapping = result.rows[0];
      this.cache.set(tenantId, mapping);
    }

    const pool = this.shards.get(mapping.shardId);
    if (!pool) {
      throw new Error(`Shard ${mapping.shardId} not configured`);
    }

    return pool;
  }

  async moveTenant(tenantId: string, newShardId: number): Promise<void> {
    // Update directory; actual data migration is a separate process
    await this.directoryPool.query(
      'UPDATE shard_directory SET shard_id = $1 WHERE tenant_id = $2',
      [newShardId, tenantId]
    );
    this.cache.delete(tenantId);
  }
}

Property	Assessment
Distribution	Full control; rebalance individual tenants without full reshards
Range queries	Depends on how directory is structured
Resharding	Easy: update directory entry, migrate data for that key
Failure isolation	Directory becomes a critical single point of failure
Implementation complexity	High; requires directory HA and cache invalidation

Failure modes: The directory service itself becomes a global bottleneck and single point of failure. You must run it with high availability (replicated, quorum-based). Also watch cache coherence: if you cache shard assignments and a migration happens, stale cache entries route writes to the wrong shard. Use short TTLs or event-based invalidation.

Geo-Based Sharding

Geo-based sharding routes data based on the geographic location of the user or the data subject. This is common in regulated industries (data residency laws) and latency-sensitive applications.

type Region = 'us-east' | 'eu-west' | 'ap-southeast';

const REGION_SHARD_MAP: Record<Region, number> = {
  'us-east': 0,
  'eu-west': 1,
  'ap-southeast': 2,
};

function inferRegionFromLocale(locale: string): Region {
  if (locale.startsWith('en-US') || locale.startsWith('en-CA')) return 'us-east';
  if (locale.startsWith('en-GB') || locale.match(/^(de|fr|es|it|nl)/)) return 'eu-west';
  return 'ap-southeast';
}

class GeoShardedRepository {
  constructor(private shards: Map<Region, Pool>) {}

  private getPool(region: Region): Pool {
    const pool = this.shards.get(region);
    if (!pool) throw new Error(`No shard for region: ${region}`);
    return pool;
  }

  async createUser(user: User, region: Region): Promise<void> {
    const pool = this.getPool(region);
    await pool.query(
      'INSERT INTO users (id, email, region, created_at) VALUES ($1, $2, $3, $4)',
      [user.id, user.email, region, user.createdAt]
    );
  }

  async findUser(userId: string, region: Region): Promise<User | null> {
    const pool = this.getPool(region);
    const result = await pool.query(
      'SELECT * FROM users WHERE id = $1',
      [userId]
    );
    return result.rows[0] ?? null;
  }
}

Property	Assessment
Distribution	Uneven; driven by user geography, not engineering choice
Range queries	Within-region excellent; cross-region expensive
Resharding	Driven by business requirements (new regions), not data growth
Failure isolation	Regional failure isolates; global queries still require fan-out
Implementation complexity	High; region must be known at write time and stored

Failure modes: Users traveling internationally or using VPNs may land on the wrong region shard if you infer region at request time. Always store the assigned region with the record and use that for lookups, not the current request’s IP. Cross-region aggregates (global user counts, analytics) require either a separate analytics layer or accepted eventual consistency.

Shard Key Selection: The Decision That Everything Else Depends On

Choosing the wrong shard key is the most expensive mistake in sharding. You can add shards, tune queries, and optimize connection pools, but rekeying data after the fact means migrating every row.

Criteria for a good shard key:

1. High cardinality. The key must have enough distinct values to distribute data across shards. Sharding on a boolean or a low-cardinality enum is almost always wrong.

2. Write distribution. Writes must spread across shards. If your shard key is created_at and you use hash sharding, you get even distribution. If you use range sharding, you get a write hotspot on the current time window.

3. Query locality. The key should align with your most frequent queries. If 90% of queries include tenant_id in the WHERE clause, tenant_id is a strong shard key candidate.

4. Immutability. Shard keys should not change after record creation. Updating a shard key means moving the record to a different shard, which is a complex, error-prone operation.

5. Inclusion in all writes. If you cannot always supply the shard key at write time, it cannot be your shard key.

For multi-tenant SaaS applications, tenant_id is almost always the right choice: it aligns query patterns, enables data isolation by tenant, and makes per-tenant migrations tractable.

Cross-Shard Queries and Joins

Sharding breaks the assumption that all data lives in one queryable space. Cross-shard queries are the tax you pay.

class CrossShardQueryExecutor {
  constructor(private shards: Pool[]) {}

  // Scatter-gather: fan out to all shards, merge results
  async globalSearch(email: string): Promise<User[]> {
    const queries = this.shards.map((pool) =>
      pool.query<User>(
        'SELECT * FROM users WHERE email = $1',
        [email]
      )
    );

    const results = await Promise.allSettled(queries);
    const users: User[] = [];

    for (const result of results) {
      if (result.status === 'fulfilled') {
        users.push(...result.rows);
      }
      // Log rejections but don't fail the whole query
    }

    return users;
  }

  // Aggregation across shards: merge partial results in application
  async globalUserCount(): Promise<number> {
    const queries = this.shards.map((pool) =>
      pool.query<{ count: string }>('SELECT COUNT(*) as count FROM users')
    );

    const results = await Promise.all(queries);
    return results.reduce((sum, r) => sum + parseInt(r.rows[0].count, 10), 0);
  }
}

Cross-shard joins are not directly possible at the database layer. Your options:

• Denormalize. Store the joined data redundantly in each shard. Acceptable when the joined data changes infrequently.
• Application-layer join. Fetch from multiple shards, join in memory. Works for small result sets. Falls apart at scale.
• Broadcast table. Replicate small, rarely-changing reference tables (e.g., countries, currencies, product catalog) to every shard. Query locally.
• Separate analytics store. Route cross-shard aggregate queries to a warehouse or OLAP store that pulls from all shards. Accepts some staleness.

Resharding and Rebalancing

At some point, you will need to add shards. If you used naive modulo hashing, nearly every record needs to move. If you used consistent hashing, only 1/N of keys need to move when you add the N+1th shard.

class ConsistentHashRouter {
  private ring = new Map<number, Pool>();
  private sortedKeys: number[] = [];
  private readonly virtualNodes = 150;

  addShard(shardId: string, pool: Pool): void {
    for (let i = 0; i < this.virtualNodes; i++) {
      const key = this.hash(`${shardId}:${i}`);
      this.ring.set(key, pool);
      this.sortedKeys.push(key);
    }
    this.sortedKeys.sort((a, b) => a - b);
  }

  getPool(key: string): Pool {
    const hash = this.hash(key);
    // Find the first node clockwise from the hash
    const index = this.sortedKeys.findIndex((k) => k >= hash);
    const nodeKey = index === -1
      ? this.sortedKeys[0]
      : this.sortedKeys[index];
    return this.ring.get(nodeKey)!;
  }

  private hash(key: string): number {
    let h = 2166136261;
    for (let i = 0; i < key.length; i++) {
      h ^= key.charCodeAt(i);
      h = (h * 16777619) >>> 0;
    }
    return h;
  }
}

For a live reshard without downtime:

1. Spin up new shards alongside existing ones.
2. Update the router to write to both old and new shards for affected keys (dual-write).
3. Backfill: read from old shards, write to new shards for all affected records.
4. Verify row counts and checksums.
5. Switch reads to new shards.
6. Stop dual-writes, decommission old shards.

This is a multi-week process on a large dataset. Plan migrations with feature flags and rollback checkpoints.

Production Considerations

Connection management. Each shard requires its own connection pool. With 8 shards and 20 connections per pool, you are holding 160 connections before your application does anything. Use connection poolers (PgBouncer) at the shard layer, not just at the application layer.

const shards: Pool[] = Array.from({ length: 8 }, (_, i) =>
  new Pool({
    host: `shard-${i}.internal`,
    database: 'app',
    max: 20,             // per-shard pool size
    idleTimeoutMillis: 30_000,
    connectionTimeoutMillis: 2_000,
  })
);

Schema migrations. Every DDL change must run against every shard. Script migrations to iterate across all shard connections and run in parallel with a timeout per shard. Track migration state per shard, not just globally, so you can retry partial failures.

async function runMigrationOnAllShards(
  shards: Pool[],
  sql: string
): Promise<void> {
  const results = await Promise.allSettled(
    shards.map(async (pool, i) => {
      const client = await pool.connect();
      try {
        await client.query(sql);
        console.log(`Shard ${i}: migration applied`);
      } finally {
        client.release();
      }
    })
  );

  const failures = results.filter((r) => r.status === 'rejected');
  if (failures.length > 0) {
    throw new Error(`Migration failed on ${failures.length} shard(s)`);
  }
}

Monitoring. Your standard database dashboard becomes N dashboards. You need per-shard metrics aggregated into a single view: query latency p99 per shard, replication lag per shard, shard size, and connection count. Alert on shard imbalance (one shard holding 3x the rows of others) as an early warning for key distribution problems.

Transaction boundaries. Distributed transactions across shards are expensive and complex. Design your data model so that operations requiring strong consistency stay within a single shard. Use eventual consistency for cross-shard state. If you need cross-shard transactions, the saga pattern is more tractable than two-phase commit at scale.

Decision Framework: Shard vs. Scale Otherwise

Before committing to sharding, exhaust these options in order:

Option	When it works	Limit
Query optimization and indexing	Query patterns are well-defined, indexes are missing	Stops working once I/O is the bottleneck
Vertical scaling	Budget allows, workload fits on one machine	Cost grows nonlinearly; ceiling exists
Read replicas	Read:write ratio is high (>4:1), read staleness is acceptable	Does not help write throughput
Connection pooling	Many short-lived connections overwhelming the DB	Not a throughput solution
Caching layer	Read-heavy, cache hit rate can be high	Does not reduce write load
Partitioning (single node)	Data has a clear time or range boundary	Still one write primary
Sharding	Multiple of the above are exhausted	Introduces distributed systems complexity

Shard when you have a write throughput problem that vertical scaling cannot solve cost-effectively, or when your working set genuinely cannot fit on a single machine. Do not shard to preempt a problem that is 18 months away. The operational complexity is real and non-trivial.

If your current architecture is a single Postgres primary with read replicas and you are under 5TB of data with no write saturation, you almost certainly do not need sharding yet. Add a partition by time range first. It gives you many of the same query performance benefits with a fraction of the operational overhead.

The Reality of Running a Sharded System

Sharding solves the scale problem. It creates a new category of operational problems that every engineer on the team must understand: knowing which shard holds a given record, debugging issues that manifest on only one shard, coordinating schema changes across shard boundaries, and handling the eventual need to reshard as data grows.

The teams that do this well build the shard routing layer as a first-class abstraction, invest in tooling that treats the fleet of shards as one logical database, and design their data model around shard boundaries from the beginning. The teams that struggle bolt sharding onto an existing schema and discover six months later that their most important query does a scatter-gather across every shard on every request.

Pick your shard key for the queries you actually run, not the queries you imagine you might run someday.