staff-engineering-skills-sharding

Sharding Trap

Sharding is nearly impossible to undo and fundamentally changes what your code can do. Before sharding, ask: have you exhausted every simpler alternative?

Do You Actually Need to Shard?

Almost certainly not. Work through this checklist first:

Only if ALL of these are insufficient should you consider sharding. A single PostgreSQL instance with proper indexing and read replicas handles far more than most people expect. If you have less than 1TB of data or fewer than 10,000 writes per second, you almost certainly don't need to shard.

What Sharding Breaks

Once data is distributed across shards, these operations become expensive or impossible:

Every feature you build must now answer: "which shard is this data on?"

Detection: Sharding-Unsafe Code

If the system is sharded or may be sharded, stop and reassess if you see:

1. JOINs across entity types -- orders JOIN products JOIN users. These tables may be on different shards. Each join may need to become a separate query + application-level join.

2. Multi-entity transactions -- BEGIN; update order; update inventory; update user; COMMIT;. If these entities are on different shards, this transaction cannot work.

3. Global unique constraints -- UNIQUE(email) only works within a single database. Cross-shard uniqueness requires a coordination layer.

4. Unscoped aggregations -- SELECT COUNT(*) FROM orders. Without a shard key in the WHERE clause, this hits every shard.

5. ORDER BY ... LIMIT without shard key -- requires fetching from all shards and merge-sorting.

Shard Key Selection

The shard key determines everything. Get it wrong and sharding makes things worse.

The shard key must be in your most common query's WHERE clause. If 80% of queries are scoped to a tenant, shard by tenant. If 80% are scoped to a user, shard by user.

Red flags in shard key selection:

• Low cardinality (country code: US gets 50% of traffic)
• Skewed distribution (first letter of name: "S" has 4x more than "Q")
• Doesn't appear in the most frequent query's WHERE clause
• Changes over time (user can move between tenants)

Patterns

Tenant-based sharding (most common for SaaS)

function getShardForTenant(tenantId: string): DatabaseConnection {
  const shardIndex = consistentHash(tenantId, shardCount);
  return shardConnections[shardIndex];
}

async function getOrders(tenantId: string, filters: OrderFilters) {
  const shard = getShardForTenant(tenantId);
  // Single-shard query -- no scatter-gather
  return shard.query(
    `SELECT * FROM orders WHERE tenant_id = $1 AND status = $2
     ORDER BY created_at DESC LIMIT $3`,
    [tenantId, filters.status, filters.limit]
  );
}

Works because almost all SaaS operations are scoped to one tenant. Cross-tenant analytics go to a separate data warehouse fed by event streaming.

Handling cross-shard queries with a read store

// Writes go to the shard
async function createOrder(tenantId: string, data: OrderInput) {
  const shard = getShardForTenant(tenantId);
  const order = await shard.insert("orders", { tenantId, ...data });

  // Async: replicate to a global read store for cross-shard queries
  await eventBus.publish("order.created", { orderId: order.id, tenantId, ...data });
  return order;
}

// Cross-shard queries hit the read store (eventually consistent)
async function globalOrderStats() {
  return analyticsDb.query(`
    SELECT DATE_TRUNC('day', created_at) as day, COUNT(*), SUM(amount)
    FROM orders_read_model GROUP BY day ORDER BY day DESC
  `);
}

Accept that cross-shard queries need a different store. The read store is eventually consistent but avoids scatter-gather.

Hot tenant isolation

const DEDICATED_TENANTS = new Map([
  ["acme-corp", dedicatedShardConnection],
  ["megacorp", dedicatedShardConnection2],
]);

function getShardForTenant(tenantId: string): DatabaseConnection {
  // Large tenants get their own shard
  const dedicated = DEDICATED_TENANTS.get(tenantId);
  if (dedicated) return dedicated;

  // Everyone else shares the pool
  const shardIndex = consistentHash(tenantId, sharedShardCount);
  return sharedShardConnections[shardIndex];
}

When one tenant generates 40% of traffic, hash-based distribution doesn't help. Move them to dedicated infrastructure.

Anti-Patterns

// Premature: "We might have millions of users someday"
// You have 50,000 users and 5GB of data. A single PostgreSQL instance
// on a $600/month machine handles 100x this. Sharding infrastructure
// will cost more in engineering time than 5 years of vertical scaling.

// Wrong shard key: sharded by user_id but dashboard queries by org_id
// Every dashboard query must scatter to ALL shards and aggregate
const results = await Promise.all(
  shards.map(shard =>
    shard.query(`SELECT count(*), sum(amount) FROM orders WHERE org_id = $1`, [orgId])
  )
);
// If the primary access pattern is per-org, shard by org, not user.

// Assumes single database: JOIN across entity types
const orderDetails = await db.query(`
  SELECT o.*, p.name, u.email
  FROM orders o
  JOIN products p ON o.product_id = p.id
  JOIN users u ON o.user_id = u.id
  WHERE o.id = $1
`, [orderId]);
// If orders, products, and users are on different shards, this doesn't work.

Related Traps

• Race Conditions -- SELECT FOR UPDATE doesn't work across shards. Distributed locking adds latency and failure modes. Cross-shard operations need saga patterns or optimistic concurrency.
• Idempotency -- the idempotency key and the data it protects must be on the same shard. Otherwise you can't atomically check for duplicates and perform the operation.
• Hot Partitions -- even with sharding, one shard can receive disproportionate traffic. Shard key distribution determines whether you've solved the problem or just renamed it.
• Consistency Models -- within a shard: strong consistency. Across shards: eventual consistency at best. Cross-shard queries see data at different points in time.
• Denormalization -- cross-shard queries often require maintaining denormalized read stores. This creates consistency obligations (see denormalization trap).