Shard Rebalancing & Data Movement: Planning, Throttling, Verification

Shard rebalancing moves key ranges between shards that already exist — because a node was added, a tenant grew, or load skewed — without changing the sharding scheme itself. It is one of the three core operations covered in the Shard Migration & Rebalancing Operations guide, alongside zero-downtime resharding workflows (which change shard count or topology wholesale) and online table repartitioning (which restructures partitions inside a single database). This page covers the full rebalance lifecycle: planning which ranges move and minimizing that set, executing throttled chunked copies, keeping reads correct while ownership is in flux, verifying with checksums and row counts, and cleaning up the source data afterward.

Problem Framing

A payments platform runs eight PostgreSQL shards keyed by account_id through the hash routing algorithms layer in its API tier. Shard 3 holds two of the platform’s largest merchant accounts and has drifted to 2.1 TB while its siblings average 900 GB. Disk forecasts give shard 3 nine weeks before it hits the storage alarm threshold, its p99 write latency is already 3× the fleet median, and autovacuum on its largest table can no longer complete inside a day.

The team cannot re-shard the whole fleet — that is a quarter-long project — and cannot repartition tables inside shard 3, because the problem is that too many keys live on one machine, not how they are arranged there. What they need is targeted movement: pick roughly 600 GB of key ranges currently owned by shard 3, copy them to the two least-loaded shards while production traffic continues, atomically switch ownership, and delete the moved rows. Every step must be resumable, rate-limited so the donor shard does not tip over mid-copy, and provably complete before anything is deleted.

That is shard rebalancing. It looks deceptively like a bulk copy, but the hard parts are all operational: choosing a move set that fixes the imbalance with minimal bytes transferred, not saturating the donor, serving correct reads while two shards both hold the same range, and never deleting a row you cannot prove arrived.

Architecture Overview

Every safe rebalance is a transition between two versions of an ownership map — the authoritative record of which shard owns which key range or virtual node. Version 1 is what routers use today. The rebalance planner computes version 2, diffs the two, and emits one movement job per range whose owner changes. Jobs copy data in throttled chunks with per-chunk checksums, a double-read window proves the destination is serving correctly, and only then does version 2 become active and version 1 retire.

Three properties make this architecture safe:

1. Ownership is versioned, never edited in place. Routers always hold a complete, immutable map version. There is no moment where a router sees a half-updated map.
2. Movement jobs are idempotent and resumable. Each job records a high-water mark per chunk; a crashed job restarts from its last checkpoint, and re-copying a chunk is harmless because writes use upsert semantics keyed on primary key and row version.
3. Deletion is a separate, final phase. Data exists on both source and destination throughout verification. Nothing is removed until version 2 is active everywhere and checksums agree.

Step 1 — Plan the Move Set

The planner’s job is to compute the smallest set of range movements that brings every shard under its target load. “Smallest” matters: every moved byte costs donor I/O, network transfer, destination write amplification, and verification time. A rebalance that moves 8% of the keyspace finishes in a night; one that moves 60% runs for a week and doubles your incident surface.

Start from measured per-range load, not per-shard totals. If you track virtual nodes (vnodes), maintain a statistics table that ties each vnode to its size and write rate:

-- Per-vnode statistics collected nightly by a fleet-wide job
CREATE TABLE IF NOT EXISTS vnode_stats (
  vnode_id      INT         NOT NULL,
  shard_id      INT         NOT NULL,
  bytes_used    BIGINT      NOT NULL,
  writes_per_s  NUMERIC     NOT NULL,
  collected_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  PRIMARY KEY (vnode_id, collected_at)
);

-- Which shards exceed 115% of the fleet mean by bytes?
WITH latest AS (
  SELECT DISTINCT ON (vnode_id) vnode_id, shard_id, bytes_used, writes_per_s
  FROM vnode_stats
  ORDER BY vnode_id, collected_at DESC
),
per_shard AS (
  SELECT shard_id, SUM(bytes_used) AS shard_bytes
  FROM latest GROUP BY shard_id
)
SELECT shard_id,
       pg_size_pretty(shard_bytes) AS size,
       ROUND(shard_bytes / AVG(shard_bytes) OVER () * 100) AS pct_of_mean
FROM per_shard
ORDER BY shard_bytes DESC;

A greedy planner then walks the overloaded shards and picks their smallest-write-rate vnodes first, assigning each to the least-loaded destination until every shard falls inside the tolerance band:

def plan_moves(vnodes, targets, tolerance=0.10):
    """vnodes: {vnode_id: (shard_id, bytes, writes_per_s)}
    targets: {shard_id: target_bytes}. Returns [(vnode_id, src, dst)]."""
    load = {s: 0 for s in targets}
    for _, (s, b, _) in vnodes.items():
        load[s] += b

    moves = []
    for src in sorted(load, key=load.get, reverse=True):
        surplus = load[src] - targets[src] * (1 + tolerance)
        if surplus <= 0:
            continue
        # Cheapest-to-move first: cold vnodes minimize double-write churn
        candidates = sorted(
            (v for v, (s, _, _) in vnodes.items() if s == src),
            key=lambda v: vnodes[v][2],
        )
        for v in candidates:
            if surplus <= 0:
                break
            dst = min(load, key=load.get)
            b = vnodes[v][1]
            moves.append((v, src, dst))
            load[src] -= b
            load[dst] += b
            surplus -= b
    return moves

Two planning rules pay for themselves repeatedly:

• Prefer cold ranges. A vnode receiving 40 writes/s generates 40 rows/s of catch-up work for the entire duration of its copy. Moving cold data rebalances storage at a fraction of the consistency cost. When the goal is spreading write load rather than bytes, invert the sort — but expect a longer double-read window.
• Cap concurrent donors and recipients. One outbound movement job per source shard and one inbound per destination keeps the throttle math simple and prevents two jobs from stacking I/O on the same node. The topology-specific arithmetic for consistent-hash rings — where adding N nodes should move only the ranges adjacent to the new vnodes — is worked through end to end in rebalancing shards after adding nodes to a consistent hash ring.

If per-range imbalance is the symptom you are chasing rather than raw disk, feed the planner from the skew signals described in partition skew detection & monitoring so the move set targets the actual hot spots.

Step 2 — Copy in Throttled Chunks

Never move a range in one transaction. A single 80 GB INSERT INTO ... SELECT across a foreign server holds a snapshot open on the donor for hours, bloats WAL on the recipient, and gives you an all-or-nothing failure mode. Chunked copy jobs read a bounded batch, write it with upsert semantics, checkpoint the high-water mark, sleep to honor the rate limit, and repeat.

import time
import psycopg2

BATCH_ROWS = 5000
MAX_ROWS_PER_SEC = 20000      # global cap per movement job
CHECKPOINT_SQL = """
    INSERT INTO move_progress (job_id, last_pk, rows_copied, updated_at)
    VALUES (%s, %s, %s, NOW())
    ON CONFLICT (job_id) DO UPDATE
      SET last_pk = EXCLUDED.last_pk,
          rows_copied = move_progress.rows_copied + EXCLUDED.rows_copied,
          updated_at = NOW()
"""

def run_movement_job(job_id, src, dst, meta, lo, hi, resume_pk=0):
    last_pk = resume_pk
    while True:
        t0 = time.monotonic()
        with src.cursor() as s:
            s.execute(
                """SELECT id, account_id, payload, row_version, updated_at
                   FROM ledger_entries
                   WHERE ring_hash > %s AND ring_hash <= %s AND id > %s
                   ORDER BY id LIMIT %s""",
                (lo, hi, last_pk, BATCH_ROWS),
            )
            rows = s.fetchall()
        if not rows:
            return last_pk
        with dst.cursor() as d:
            d.executemany(
                """INSERT INTO ledger_entries
                     (id, account_id, payload, row_version, updated_at)
                   VALUES (%s, %s, %s, %s, %s)
                   ON CONFLICT (id) DO UPDATE
                     SET payload = EXCLUDED.payload,
                         row_version = EXCLUDED.row_version,
                         updated_at = EXCLUDED.updated_at
                   WHERE ledger_entries.row_version < EXCLUDED.row_version""",
                rows,
            )
        dst.commit()
        last_pk = rows[-1][0]
        with meta.cursor() as m:
            m.execute(CHECKPOINT_SQL, (job_id, last_pk, len(rows)))
        meta.commit()
        # Throttle: pad the loop so effective rate stays under the cap
        elapsed = time.monotonic() - t0
        time.sleep(max(0.0, len(rows) / MAX_ROWS_PER_SEC - elapsed))

The details that separate a production job from a script:

• Keyset pagination, not OFFSET. WHERE id > last_pk ORDER BY id LIMIT n stays O(batch) regardless of how deep into the range the job is. OFFSET degrades linearly and re-reads skipped rows on every batch.
• Version-guarded upserts. The WHERE row_version < EXCLUDED.row_version clause makes re-copying safe even if a fresher row already landed on the destination through catch-up replication or a retried chunk.
• A feedback throttle beats a fixed one. The static cap above is the floor; a production job also samples donor replica lag and p99 latency every few batches and halves its rate when either breaches a threshold. Wire those samples into the alert rules from replication lag & capacity alerting so a saturated donor pages a human before customers notice.
• Catch-up for hot ranges. For ranges taking live writes, the bulk copy captures a point-in-time snapshot and a second phase drains changes recorded after the snapshot — either from an updated_at-indexed delta query run in the same chunked loop, or from logical decoding on the donor. Iterate the delta pass until the remaining lag is a few seconds, then hold the double-write or double-read window while the tail drains.

Step 3 — Keep Reads Consistent During Movement

While a range is mid-flight, two shards hold overlapping data and neither is trivially authoritative. Two mechanisms keep queries correct: a versioned ownership map and a double-read window.

Versioned ownership map. Routers never consult a mutable table of assignments. They load an immutable map version — from etcd, Redis, or a shard_map table with a version column — and every query is executed against the owner as of the version the router currently holds. Publishing version 2 is a single atomic write of a new version pointer:

-- Ownership map storage: append-only versions, one atomic pointer
CREATE TABLE shard_map_versions (
  version     INT  PRIMARY KEY,
  created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE shard_map_entries (
  version   INT  NOT NULL REFERENCES shard_map_versions(version),
  range_lo  BIGINT NOT NULL,
  range_hi  BIGINT NOT NULL,
  shard_id  INT  NOT NULL,
  PRIMARY KEY (version, range_lo)
);

CREATE TABLE shard_map_active (
  singleton  BOOL PRIMARY KEY DEFAULT TRUE CHECK (singleton),
  version    INT NOT NULL REFERENCES shard_map_versions(version)
);

-- The cutover is one row update; routers poll or subscribe to it
UPDATE shard_map_active SET version = 2;

During the transition there is an unavoidable interval where some routers hold v1 and others hold v2. Keep it correct by ordering the phases so that both answers are right: the destination already has a complete, caught-up copy before v2 is published, and the source keeps its copy (and keeps receiving forwarded writes, if the range is hot) until every router acknowledges v2. Routers should report their loaded version as a metric; the cutover controller waits until min(router_version) >= 2 before allowing the delete phase.

Double reads. For the window between copy completion and cutover, the router (or a verification sidecar) issues reads for moved ranges against both shards and compares results:

def double_read(key, src_conn, dst_conn, metrics):
    a = fetch_row(src_conn, key)   # still authoritative
    b = fetch_row(dst_conn, key)
    if row_fingerprint(a) != row_fingerprint(b):
        metrics.incr("rebalance.double_read.mismatch",
                     tags={"range": owning_range(key)})
    return a  # serve from the current owner until cutover

Sample double reads (1–5% of traffic is typical) rather than duplicating every read if latency budgets are tight. The rule is absolute either way: the mismatch counter must be zero over a full sampling window before the version flip. A nonzero counter means catch-up is still draining or the copy missed rows — flip nothing until it is explained.

Writes during the window follow one of two disciplines. Either the source remains the only writer and the delta pass keeps draining (simpler, needs a final short write pause of a few seconds at flip time), or writes are applied to both shards with the version-guarded upsert (no pause, but requires idempotent write paths). Pick one per rebalance; mixing them is how ranges end up with rows on the wrong side.

Step 4 — Verify with Checksums and Row Counts

Row counts catch missing chunks; checksums catch corrupt or stale ones. Run both per moved range, on the same snapshot boundary (after writes to the range are quiesced or after the delta pass reports zero lag):

-- Run identically on source and destination for each moved range
SELECT
  COUNT(*)                                            AS row_count,
  SUM(hashtext(id::text || ':' || row_version::text)) AS content_sum,
  MD5(STRING_AGG(md5(t.*::text), '' ORDER BY id))     AS strict_digest
FROM ledger_entries t
WHERE ring_hash > 3221225472 AND ring_hash <= 3251200000;

The SUM(hashtext(...)) form is order-independent and cheap enough to run per chunk during the copy (a per-chunk checksum cadence catches a bad batch within minutes instead of at the end of a 12-hour job). The STRING_AGG digest is the strict end-of-job check — it hashes full row images in key order and will catch any divergence, at the cost of a full range scan. On very large ranges, sample the strict digest over a deterministic subset (WHERE id % 97 = 0) and rely on the running per-chunk sums for full coverage.

Record every verification result in the movement job’s metadata table. The delete phase must refuse to start unless the latest verification row for the range shows source_count = dest_count, matching digests, and a double-read mismatch count of zero.

Step 5 — Clean Up Moved Data

After v2 is active on every router and verification has passed, the source rows are garbage — but they are garbage sitting in a live production table, so deletion gets the same throttling discipline as the copy:

-- Batched, throttled delete driven by an external loop
DELETE FROM ledger_entries
WHERE ctid IN (
  SELECT ctid FROM ledger_entries
  WHERE ring_hash > 3221225472 AND ring_hash <= 3251200000
  LIMIT 5000
);
-- loop with a sleep until 0 rows affected, then:
VACUUM (ANALYZE) ledger_entries;

Keep a deliberate grace period — 24 to 72 hours between cutover and delete is common — so an unexpected v2 problem can be rolled back to v1 with the source data intact. During the grace period, the source copy is stale the moment the first post-cutover write lands on the destination, so a rollback after that point needs the reverse delta (destination back to source) before v1 can be re-activated. Once the grace period ends and deletion completes, reclaim space with vacuum and confirm the donor shard’s size and latency actually dropped — the whole point of the exercise.

Configuration Reference

Operational Contrast

Rebalancing is the right tool when the assignment of ranges to healthy shards is wrong but the sharding scheme is fine. Its defining constraints are minimal movement and a long-lived dual-copy window managed by versioned ownership.

Zero-downtime resharding workflows solve a different problem: the shard topology itself changes — going from 8 to 16 shards, splitting a hot shard, or changing the shard key. Resharding typically rewrites most or all of the keyspace, leans on logical replication rather than application-level copy jobs, and is planned as a project with a coordinated cutover rather than a background maintenance activity. If your plan-the-move-set step keeps concluding that more than about a third of the keyspace must move, you have outgrown rebalancing and should treat the work as a reshard.

Online table repartitioning operates one level down: data stays on the same database node, but its partition layout changes — converting an unpartitioned table to a partitioned one or changing partition boundaries. No ownership map is involved and no bytes cross the network; the challenges are DDL locking, trigger-based dual writes, and planner behavior instead of throttling and cross-node verification. Fleets often chain the two: repartition a table so its partitions align with vnode boundaries, precisely so future rebalances can move whole partitions with DETACH/ATTACH instead of row-by-row copies.

Failure Modes

Common Mistakes

• Recomputing placement from scratch instead of diffing maps. A planner that reassigns every range “optimally” produces a near-total data migration. Always compute the move set as the difference between the current map and the target map, and audit total moved bytes against the theoretical minimum before starting.
• Deleting on copy completion instead of verification completion. “The job exited 0” is not proof of completeness — a resumed job with a wrong checkpoint or a silently failed batch exits 0 too. Deletion gates on recorded row counts, digests, and a clean double-read window, never on job status.
• Throttling only the copy, not the cleanup. A DELETE of 200 million rows generates WAL, index maintenance, and vacuum debt comparable to the copy itself. Unthrottled cleanup has taken down more donors than unthrottled copy, because it runs when everyone has stopped paying attention.
• Letting the ownership map drift from reality. Manual hotfixes — “just point range X at shard 5 for now” — that bypass the versioned map leave routers and planners reasoning from fiction. Every ownership change, including emergency ones, must be a new map version, or the next rebalance will be planned against wrong data.

FAQ

Related

• Shard Migration & Rebalancing Operations — parent guide covering the full lifecycle of moving and reshaping sharded data
• Rebalancing Shards After Adding Nodes to a Consistent Hash Ring — concrete walkthrough of the vnode arithmetic, throttled pre-copy, and atomic ring flip for the most common rebalance trigger
• Zero-Downtime Resharding Workflows — when the topology itself must change and movement is no longer minimal
• Online Table Repartitioning — restructuring partitions inside a single node so future rebalances can move whole partitions instead of rows