Partitioning and sharding

• Shard when one primary cannot take the writes, or cannot hold the data.
• Each shard owns a subset of keys. A router maps a key to its shard.
• The placement rule decides how much data moves when the cluster grows.

I never place keys with hash mod N on a cluster that grows. Adding one node would move most of the data.

Fixed slots and the ring both move only the new node's share. Slots keep an explicit map; a ring derives the map from token positions.

I use ranges when queries scan ordered keys, and hashing into slots or a ring when writes must spread evenly.

I pick the key that the most common query names. Then I check that no value takes a large share of writes, and that new data spreads too.

The router finds the shard from a cached map. A query with the key touches one shard; a query without it touches all of them.

On one Postgres I partition tables and let pruning skip partitions. Across machines I use slots or a ring, and I move data with logical replication or double writes.

build(nodes, vnodes):                      // once per membership change
  FOR EACH node IN nodes:
    FOR i IN 0 .. vnodes-1:
      put token hash(node + "#" + i) on the ring1
  sort tokens by position

owner(key):
  h = hash(key)
  t = first token at or after h2             // binary search
  IF there is none: t = the first token     // wrap around3
  RETURN t.node

add(node):
  put the vnodes tokens of node on the ring
  keys just before each new token move to node4

// Ring is a consistent hash ring. Each node owns vnodes points (tokens) on the ring; a key
// belongs to the first token at or after its hash, wrapping at the end.
type Ring struct {
  tokens []Token // sorted by Pos
}

// Token is one virtual node: a point on the ring and the node that owns it.
type Token struct {
  Pos  uint64
  Node string
}

// NewRing places vnodes tokens for each node. A token's position depends only on its node and
// index, so adding a node never moves another node's tokens.
func NewRing(nodes []string, vnodes int) *Ring {
  r := &Ring{}
  for _, n := range nodes {
    r.tokens = append(r.tokens, NodeTokens(n, vnodes)...)
  }
  sort.Slice(r.tokens, func(i, j int) bool { return r.tokens[i].Pos < r.tokens[j].Pos })
  return r
}

// NodeTokens returns the tokens of one node.
func NodeTokens(node string, vnodes int) []Token {
  out := make([]Token, vnodes)
  for v := range out {
    out[v] = Token{Pos: Hash(fmt.Sprintf("%s#%d", node, v)), Node: node}
  }
  return out
}

// Owner implements Placer: binary search for the first token at or after the key's hash.
func (r *Ring) Owner(key string) string {
  h := Hash(key)
  i := sort.Search(len(r.tokens), func(i int) bool { return r.tokens[i].Pos >= h })
  if i == len(r.tokens) {
    i = 0 // past the last token: wrap to the first
  }
  return r.tokens[i].Node
}

A key belongs to the first token clockwise from its hash. A new node only takes the arcs just before its own tokens.

Going from 4 to 5 nodes, hash mod N moves about 80% of the keys. Slots and a ring with enough tokens move about 20%, the new node's fair share.

Six nodes, any subset, five token counts: 315 rings, each simulated over 20,000 keys. Dots are 48 of those keys.

With one token per node the arcs are uneven and a new node takes keys from one neighbour. With a few hundred tokens each node owns many small arcs, so load and moves both even out.

A timestamp as the key sends every new write to the newest range. I hash the source first and keep time inside the shard.

write(key, value):
  IF key is hot:
    key = key + "#" + (count MOD k1)         // k sub-keys
  shard(key).write(value)

read(key):
  IF key is hot:
    RETURN merge(2shard(key + "#" + i).read FOR i IN 0 .. k-1)
  RETURN shard(key).read

// Salt spreads the writes of one hot key over k sub-keys ("user:0#0" to "user:0#k-1"), round
// robin. Each sub-key hashes to its own shard. A read of the hot key must read all k sub-keys.
func Salt(writes []string, hot string, k int) []string {
  out := make([]string, len(writes))
  next := 0
  for i, w := range writes {
    if w != hot {
      out[i] = w
      continue
    }
    out[i] = fmt.Sprintf("%s#%d", hot, next%k)
    next++
  }
  return out
}

Sub-keys hash like any key, so two can land on one shard. The whisker shows that spread.

One key with a quarter of the writes made its shard take 34.4% of all writes. Split into 8 sub-keys, the hottest shard took about 17.3%.

-- One logical table, split into 8 partitions by a hash of tenant_id.
CREATE TABLE events (
  tenant_id  bigint      NOT NULL,
  event_id   bigint      NOT NULL,
  created_at timestamptz NOT NULL,
  body       text        NOT NULL,
  PRIMARY KEY (tenant_id, event_id)2
) PARTITION BY HASH (tenant_id)1;

CREATE TABLE events_p0 PARTITION OF events
  FOR VALUES WITH (MODULUS 8, REMAINDER 03);
CREATE TABLE events_p1 PARTITION OF events
  FOR VALUES WITH (MODULUS 8, REMAINDER 1);
-- ... events_p2 to events_p7

A query without the shard key goes to every shard. For a common lookup I keep a global index or put the shard number in the ID.

I choose the key so that one transaction touches one shard. When it cannot, I run a saga with compensations, not two-phase commit.

The lab error, verbatim: "unique constraint on partitioned table must include all partitioning columns".

A unique index works only inside one shard. I generate IDs that are unique by design, and I claim unique values like emails in a table sharded by that value.

move(range, old, new):
  1  turn on double writes:
       write old shard                      // version = version + 1
       write new shard IF its version is older1
  2  backfill: FOR EACH batch of rows IN old:
       copy to new, SKIP rows new already has2
  3  verify: rows that differ == 03
  4  switch reads of the range to new4
  5  stop writes to old, then delete the range from old

// Write changes one account during a move: old shard first, then the new shard. The version
// guard on the new shard makes the two writes safe in any order across writers.
func Write(ctx context.Context, db *pgxpool.Pool, id, delta int64) error {
  var balance, version int64
  if err := db.QueryRow(ctx, reshard["write_old"], id, delta).Scan(&balance, &version); err != nil {
    return fmt.Errorf("write old shard, account %d: %w", id, err)
  }
  if _, err := db.Exec(ctx, reshard["write_new"], id, balance, version); err != nil {
    return fmt.Errorf("write new shard, account %d: %w", id, err)
  }
  return nil
}

// Backfill copies the old shard to the new one in batches of size, in id order, while the
// service keeps double writing.
func Backfill(ctx context.Context, db *pgxpool.Pool, size int) error {
  var after int64
  for {
    batch, err := ReadBatch(ctx, db, after, size)
    if err != nil {
      return err
    }
    if len(batch) == 0 {
      return nil
    }
    if err := CopyBatch(ctx, db, batch, false); err != nil {
      return err
    }
    after = batch[len(batch)-1].ID
  }
}

-- The service writes the old shard first. The version counts every change to the row.
UPDATE accounts_old
SET balance = balance + $2, version = version + 1
WHERE id = $1
RETURNING balance, version;

-- Then it writes the same row to the new shard. An older version never replaces a newer one.
INSERT INTO accounts_new (id, balance, version) VALUES ($1, $2, $3)
ON CONFLICT (id) DO UPDATE
SET balance = excluded.balance, version = excluded.version
WHERE accounts_new.version < excluded.version;

-- The backfill skips a row the new shard already has. A double write put it there, and it is newer.
INSERT INTO accounts_new (id, balance, version)
SELECT * FROM unnest($1::bigint[], $2::bigint[], $3::bigint[])
ON CONFLICT (id) DO NOTHING;

-- Rows that differ between the shards, or exist on one side only.
SELECT count(*)
FROM accounts_old o
FULL JOIN accounts_new n USING (id)
WHERE (o.balance, o.version) IS DISTINCT FROM (n.balance, n.version);

I double write with a version guard, backfill rows the new shard does not have, verify, and only then switch the map. The backfill never overwrites a row.

Demand example: 1 billion events a day is 1,000,000,000 / 86,400 ≈ 11,574 a second. Shard capacity is the lab measurement times the shard count; real shards scale a little below that. Postgres 16 on an 8-core laptop, 32 clients.

I add replicas for reads, a bigger machine and partitions for size. I shard only when one primary cannot take the writes, and I pick the key from the main query.