Shard Migration & Rebalancing Operations

Every partitioned system eventually has to move data it is actively serving: a shard grows past its scaling ceiling, a hot tenant needs its own hardware, or an unpartitioned table finally becomes too large to manage in place. This guide is the operations discipline for those moves — planning, executing, and verifying data movement between partitions and shards while production traffic continues. It covers the full lifecycle in three sub-topics: Zero-Downtime Resharding Workflows for splitting and merging live shards, Online Table Repartitioning for converting monolithic tables to partitioned layouts, and Shard Rebalancing & Data Movement for redistributing keys after topology changes.

The unifying model across all three is a five-phase pipeline: mirror new writes, backfill history, verify parity, cut over routing, then clean up. Everything else — tool choice, throttling, monitoring — is detail inside those phases.

Why Shards Move: Migration Triggers

Data movement is expensive and risky, so every migration should trace back to a concrete operational trigger. Four dominate in production:

1. A shard approaches a hard ceiling. Disk, IOPS, connection count, or vacuum throughput on one node can no longer keep up. The economics of splitting versus scaling up are covered in the Scaling Limits & Cost Tradeoffs guide linked above; once vertical headroom is gone, a split is the only option.
2. Key skew concentrates load. One tenant or region grows to dominate a shard that was balanced at design time. The fix is moving that keyspace to dedicated hardware — a targeted migration, not a full reshuffle.
3. Topology changes. Adding nodes to absorb growth means a fraction of every existing shard’s keys must relocate. With consistent hashing this is bounded; with modulo routing it is a near-total rewrite.
4. A monolithic table needs a partitioned layout. The table was never partitioned, and now VACUUM takes twelve hours and every index rebuild is an outage risk. Converting it online is its own discipline, covered in the Online Table Repartitioning sub-guide.

Quantify the trigger before committing. This query gives the per-shard baseline you will compare against after the move:

-- Run on each shard: size, dead-tuple pressure, and top keyspace consumers
SELECT inet_server_addr()::text                              AS shard,
       pg_size_pretty(pg_database_size(current_database()))  AS db_size,
       (SELECT round(sum(n_dead_tup)::numeric /
                     NULLIF(sum(n_live_tup), 0) * 100, 1)
        FROM pg_stat_user_tables)                            AS dead_pct;

-- Which keyspace is driving growth on this shard (last 30 days)
SELECT region, count(*) AS rows_30d,
       pg_size_pretty(sum(pg_column_size(o.*))::bigint) AS approx_bytes
FROM   orders o
WHERE  placed_at >= now() - interval '30 days'
GROUP  BY region
ORDER  BY rows_30d DESC;

If one key value accounts for more than roughly 40 % of a shard’s recent writes, plan a targeted keyspace move. If growth is uniform, plan a topology expansion and rebalance instead — the two have very different blast radii.

Migration Strategy Taxonomy

Four mechanisms cover essentially every production migration on PostgreSQL and MySQL. The choice hinges on engine, whether the move crosses instances, and how much write pause you can tolerate.

Logical replication

The default for cross-instance moves on PostgreSQL. The source publishes row changes; the target subscribes, performs an initial copy, then applies the change stream continuously. PostgreSQL 15 row filters let you publish only the keyspace that is moving:

-- Source shard (PostgreSQL 15+): publish only the migrating keyspace
ALTER TABLE orders REPLICA IDENTITY FULL;  -- or ensure a PK covers updates/deletes
CREATE PUBLICATION pub_move_eu FOR TABLE orders WHERE (region = 'eu');

-- Target shard: subscribe; copy_data performs the backfill automatically
CREATE SUBSCRIPTION sub_move_eu
  CONNECTION 'host=pg-shard-2.internal dbname=commerce user=replicator'
  PUBLICATION pub_move_eu
  WITH (copy_data = true, streaming = 'on');

On PostgreSQL 10–14 there are no row filters, so either replicate the whole table and delete the non-migrating rows on the target after cutover, or publish a partition child directly if the table is already partitioned by the moving dimension. The complete split procedure, including slot management and multi-target fan-out, is the subject of the Zero-Downtime Resharding Workflows sub-guide.

Trigger-based dual-write

For same-instance moves — classically, migrating a monolithic table into a new partitioned table — a trigger mirrors every write while a batch job backfills history:

-- New partitioned target lives alongside the old table
CREATE OR REPLACE FUNCTION mirror_orders_write() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  IF TG_OP = 'INSERT' THEN
    INSERT INTO orders_partitioned VALUES (NEW.*)
    ON CONFLICT (order_id, placed_at) DO NOTHING;
  ELSIF TG_OP = 'UPDATE' THEN
    UPDATE orders_partitioned SET (status, total_usd, updated_at) =
      (NEW.status, NEW.total_usd, NEW.updated_at)
    WHERE order_id = NEW.order_id AND placed_at = NEW.placed_at;
  ELSIF TG_OP = 'DELETE' THEN
    DELETE FROM orders_partitioned
    WHERE order_id = OLD.order_id AND placed_at = OLD.placed_at;
  END IF;
  RETURN NULL;
END $$;

CREATE TRIGGER trg_mirror_orders
AFTER INSERT OR UPDATE OR DELETE ON orders
FOR EACH ROW EXECUTE FUNCTION mirror_orders_write();

The ON CONFLICT DO NOTHING on insert makes the trigger idempotent against the backfill: whichever writes a given row first wins, and the other becomes a no-op. That idempotence is what lets the trigger and the batch copy run concurrently without coordination.

gh-ost / pt-online-schema-change

On MySQL, ALTER TABLE ... PARTITION BY rebuilds the table under a lock, so online tooling is mandatory at scale. gh-ost creates a ghost copy with the new layout, backfills it in chunks, and tails the binlog to stay current:

gh-ost \
  --host=mysql-shard-1.internal --database=commerce --table=orders \
  --alter="PARTITION BY RANGE COLUMNS (placed_at) (
      PARTITION p2026q1 VALUES LESS THAN ('2026-04-01'),
      PARTITION p2026q2 VALUES LESS THAN ('2026-07-01'),
      PARTITION pmax    VALUES LESS THAN (MAXVALUE))" \
  --chunk-size=1000 \
  --max-load="Threads_running=25" \
  --critical-load="Threads_running=64" \
  --throttle-control-replicas="mysql-shard-1-replica.internal" \
  --max-lag-millis=1500 \
  --postpone-cut-over-flag-file=/tmp/orders.postpone \
  --execute

The --postpone-cut-over-flag-file is the operational safety valve: gh-ost finishes the copy and then idles, applying binlog events, until you delete the flag file — so the atomic table swap happens on your schedule, not the tool’s.

Dump-and-load

pg_dump | pg_restore (or mysqldump) remains legitimate when the moving keyspace is small enough that a write freeze of a few minutes is acceptable, and it is the standard way to seed a target before switching to replication for the delta:

pg_dump -h pg-shard-2.internal -d commerce \
        --table=orders --data-only -Fc \
  | pg_restore -h pg-shard-5.internal -d commerce \
               --data-only --jobs=4 --no-owner

Never use dump-and-load alone for anything above ~50 GB or any keyspace with continuous writes; the freeze window grows linearly and the rollback story is poor.

Routing During the Migration Window

While data lives in two places, the routing layer is the component that decides correctness. The migration state machine belongs next to the shard map — the same structure your hash routing algorithms already consult — so every application instance sees a consistent phase.

# Migration-aware routing: the shard map carries per-keyspace migration state
PHASES = ("dual_write", "backfill", "verify", "cutover", "cleanup")

class MigrationAwareRouter:
    def __init__(self, shard_map, migration_registry):
        self.shard_map = shard_map            # key -> shard resolution
        self.migrations = migration_registry  # shard/keyspace -> Migration

    def route_write(self, key):
        source = self.shard_map.resolve(key)
        m = self.migrations.lookup(source, key)
        if m is None:
            return [source]
        if m.phase in ("dual_write", "backfill", "verify"):
            return [source, m.target]   # source is authoritative; target mirrored
        if m.phase in ("cutover", "cleanup"):
            return [m.target]           # target is now authoritative
        return [source]

    def route_read(self, key):
        source = self.shard_map.resolve(key)
        m = self.migrations.lookup(source, key)
        if m and m.phase in ("cutover", "cleanup"):
            return m.target
        return source                    # reads stay on source until cutover

Three rules keep this correct:

• The source stays authoritative until cutover. Reads never touch the target during dual-write, because the backfill has not finished and the target is known-incomplete. Serving reads from both sides “for testing” is how phantom missing-row bugs enter production.
• Write the source first, mirror second. If the mirrored write fails, enqueue it for replay and alert — do not fail the user request, and do not silently drop it. A dead-letter queue of failed mirror writes is a first-class migration artifact.
• Phase transitions are one-way and centrally stored. Keep the phase in a shard_map table or an etcd key with a version number; application instances must refuse to act on a phase older than one they have already seen.

Note that during a shard split, queries that previously hit one shard may temporarily fan out across two. If your workload includes multi-key reads, review Cross-Partition Querying & Aggregation Strategies before the split — scatter-gather behaviour during the mixed window is where most latency regressions surface.

Operational Configuration

Migrations fail operationally more often than logically: a replication slot fills a disk, an unthrottled backfill starves OLTP traffic, or the target applies changes slower than the source produces them. Configure both sides before opening the stream.

# postgresql.conf — source shard
wal_level                 = logical
max_wal_senders           = 10
max_replication_slots     = 10
max_slot_wal_keep_size    = 200GB    # hard cap: a stalled subscriber cannot fill the disk
wal_sender_timeout        = 60s

# postgresql.conf — target shard
max_logical_replication_workers = 8
max_sync_workers_per_subscription = 2   # parallel initial-copy workers
maintenance_work_mem      = 2GB          # faster index builds after the copy
autovacuum_vacuum_scale_factor = 0.02    # keep up with backfill churn

max_slot_wal_keep_size deserves emphasis. A replication slot pins WAL on the source until the subscriber consumes it; if the target stalls for a weekend, an uncapped slot will fill the source’s disk and take down the healthy shard. The cap sacrifices the migration (the slot is invalidated) to protect production — the correct trade in every case.

Encode the backfill policy as configuration rather than constants in a script, so the throttle can be tuned mid-flight without redeploying:

# migration-job.yaml — declarative migration run book
migration:
  id: move-eu-orders-to-shard5
  source: pg-shard-2.internal
  target: pg-shard-5.internal
  table: orders
  keyspace: "region = 'eu'"
  backfill:
    batch_rows: 5000            # rows per INSERT ... SELECT batch
    pause_ms: 50                # sleep between batches
    max_replica_lag_seconds: 10 # auto-pause if source replicas fall behind
    max_source_cpu_pct: 60      # auto-pause above this
  verify:
    checksum_buckets: 64
    recheck_recent_hours: 24    # continuously re-verify the hot tail
  cutover:
    max_write_pause_ms: 3000    # abort cutover if the pause would exceed this
    rollback_window_hours: 48   # reverse replication retained this long

The two auto-pause conditions matter most. Backfill traffic is the lowest-priority work in the system; it should yield to OLTP latency and to replica lag automatically, not when a human notices a dashboard.

Monitoring & Observability

A migration in flight needs its own dashboard. Wire these signals into the same alerting stack described in Partition Monitoring & Failover Automation before phase 1 begins, because the failure modes — slot bloat, apply lag, silent drift — all develop while everything looks fine.

Catalog queries

Initial copy progress, per table, on the target:

-- Subscription sync state: i = initializing, d = copying data,
-- f = finished copy, s = synchronized, r = ready (streaming)
SELECT srrelid::regclass AS table_name, srsubstate AS state
FROM   pg_subscription_rel
ORDER  BY srsubstate;

Slot retention on the source — the disk-full early warning:

SELECT slot_name,
       active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn))
         AS retained_wal
FROM   pg_replication_slots
WHERE  slot_name LIKE 'sub_move%';

Apply lag on the target, in bytes and seconds:

SELECT subname,
       pg_size_pretty(pg_wal_lsn_diff(latest_end_lsn, received_lsn)) AS recv_gap,
       extract(epoch FROM now() - latest_end_time)                   AS apply_lag_s
FROM   pg_stat_subscription
WHERE  subname = 'sub_move_eu';

PromQL alerts

# Replication slot retaining more than 50 GB of WAL on the source
max by (slot_name) (pg_replication_slot_retained_wal_bytes) > 50e9

# Subscription apply lag above 60 s for 10 minutes
max by (subname) (pg_subscription_apply_lag_seconds) > 60

# Backfill starving OLTP: p99 write latency on the source doubles vs 1 h ago
histogram_quantile(0.99, sum by (le) (rate(pg_query_duration_seconds_bucket{shard="pg-shard-2",op="write"}[5m])))
  > 2 * histogram_quantile(0.99, sum by (le) (rate(pg_query_duration_seconds_bucket{shard="pg-shard-2",op="write"}[5m] offset 1h)))

# Dead-letter queue of failed mirror writes is non-empty
sum(migration_mirror_dlq_depth{migration="move-eu-orders-to-shard5"}) > 0

Set the paging thresholds below the hard limits: page at 50 GB retained WAL when max_slot_wal_keep_size is 200 GB, and at 60 s apply lag when your cutover requires lag ≈ 0. The gap between the page and the ceiling is your time to react. The mirror-write dead-letter alert is non-negotiable — a single unreplayed mirror write is a parity violation that checksums will catch hours later at much higher diagnostic cost.

The Zero-Downtime Migration Workflow

The following numbered workflow moves the region = 'eu' keyspace from pg-shard-2 to a new pg-shard-5 using logical replication. It is the canonical shape; the sub-guides linked above adapt it for splitting hot shards and for same-instance repartitioning.

1. Provision the target and copy the schema. Data replication does not carry DDL, so the schema (including indexes, constraints, and roles) is copied explicitly, and DDL is frozen on the source for the duration:

pg_dump -h pg-shard-2.internal -d commerce \
        --schema-only --table=orders \
  | psql -h pg-shard-5.internal -d commerce

# Declare the DDL freeze where your team will see it
echo "DDL FREEZE: orders (move-eu-orders-to-shard5) until cutover" \
  >> /etc/motd

2. Open the publication and subscription (the CREATE PUBLICATION / CREATE SUBSCRIPTION pair from the taxonomy section above). The subscription’s copy_data = true starts the backfill immediately using max_sync_workers_per_subscription parallel workers.

3. Watch the backfill and let the throttle work. Poll sync state until every table reports r, and keep an eye on source impact:

while true; do
  state=$(psql -h pg-shard-5.internal -d commerce -Atc \
    "SELECT string_agg(DISTINCT srsubstate, ',') FROM pg_subscription_rel")
  echo "$(date -Is) sync_state=${state}"
  [ "$state" = "r" ] && break
  sleep 60
done

4. Synchronise sequences. Logical replication never advances sequences on the target. Set them past the source’s current values with a safety margin:

-- On the target, after the copy reaches streaming state
SELECT setval('orders_order_id_seq',
              (SELECT max(order_id) FROM orders) + 100000);

The + 100000 offset absorbs writes that land on the source between this statement and cutover. Gaps in a sequence are free; collisions are an outage.

5. Verify parity with bucketed checksums. Row counts catch missing data but not corrupted or stale rows. Hash-bucket digests catch both and parallelise well:

-- Run identically on source and target; diff the outputs
SELECT mod(abs(hashtext(order_id::text)), 64)              AS bucket,
       count(*)                                            AS rows,
       md5(string_agg(md5(o::text), '' ORDER BY order_id)) AS digest
FROM   orders o
WHERE  region = 'eu'
GROUP  BY 1
ORDER  BY 1;

psql -h pg-shard-2.internal -d commerce -Atc "$CHECKSUM_SQL" > /tmp/src.sum
psql -h pg-shard-5.internal -d commerce -Atc "$CHECKSUM_SQL" > /tmp/dst.sum
diff /tmp/src.sum /tmp/dst.sum && echo "PARITY OK"

A mismatched bucket confines the investigation to 1/64th of the keyspace. Expect transient mismatches in buckets containing rows written in the last few seconds; a bucket that stays mismatched across two runs ten minutes apart is real drift. Re-verify the recent-data window continuously until cutover.

6. Cut over with a bounded write pause. Pause the pooler, wait for the target to confirm it has applied everything, flip the routing map, resume. The pause is typically one to three seconds:

# 1. Pause new transactions to the moving keyspace
psql -h pgbouncer.internal -p 6432 -U admin pgbouncer -c "PAUSE commerce_shard2;"

# 2. Wait for the target to reach LSN parity with the source
SRC_LSN=$(psql -h pg-shard-2.internal -d commerce -Atc \
  "SELECT pg_current_wal_lsn()")
until psql -h pg-shard-5.internal -d commerce -Atc \
  "SELECT received_lsn >= '${SRC_LSN}'::pg_lsn FROM pg_stat_subscription
   WHERE subname='sub_move_eu'" | grep -q t; do sleep 0.2; done

# 3. Flip the authoritative routing entry (versioned, single row)
psql -h pg-meta.internal -d topology -c \
  "UPDATE shard_map SET shard='pg-shard-5', version=version+1
   WHERE keyspace='eu';"

# 4. Resume traffic — now routed to the target
psql -h pgbouncer.internal -p 6432 -U admin pgbouncer -c "RESUME commerce_shard2;"

If step 2 does not converge within max_write_pause_ms, resume immediately and abort the cutover attempt — the migration stays in verify phase and nothing is lost. A cutover you can abandon cheaply is a cutover you can attempt calmly.

7. Establish reverse replication for the rollback window. Immediately after cutover, publish the keyspace from the target back to the source, so a rollback is a routing flip rather than a data recovery:

-- On pg-shard-5 (now authoritative)
CREATE PUBLICATION pub_rollback_eu FOR TABLE orders WHERE (region = 'eu');
-- On pg-shard-2: subscribe with copy_data = false — history already matches
CREATE SUBSCRIPTION sub_rollback_eu
  CONNECTION 'host=pg-shard-5.internal dbname=commerce user=replicator'
  PUBLICATION pub_rollback_eu
  WITH (copy_data = false);

8. Clean up after the soak period. Once the rollback window (48 hours in the run book above) passes with clean error budgets, tear down in strict order — replication first, then data:

-- On the target: drop the forward subscription (drops its slot on the source)
DROP SUBSCRIPTION sub_move_eu;
-- On the source: drop reverse replication, then delete migrated rows in batches
DROP SUBSCRIPTION sub_rollback_eu;
DELETE FROM orders WHERE region = 'eu'
  AND ctid IN (SELECT ctid FROM orders WHERE region = 'eu' LIMIT 10000);
-- repeat the batched DELETE until 0 rows, then reclaim space
VACUUM (ANALYZE) orders;

Rollback planning

Write the rollback procedure before phase 1, and rehearse it once on staging. It has exactly three moves: flip the shard_map entry back to the source (reverse replication has kept it current), re-enable the forward mirror if the migration will be retried, and file the drift report from whatever checksum bucket triggered the rollback. Define the rollback triggers numerically in advance — for example, any parity mismatch after cutover, target p99 latency above 1.5× the source baseline for 15 minutes, or the dead-letter queue exceeding zero — so the on-call engineer executes a decision that was already made rather than debating one at 03:00.

Rebalancing After Topology Changes

Splitting one hot shard is a scalpel; rebalancing after adding nodes is a conveyor belt — many small keyspace moves executed with the same five-phase pipeline, one keyspace at a time. With consistent hashing, adding one node to an N-node ring relocates roughly 1/(N+1) of all keys, and virtual nodes ensure the donated keys come evenly from every existing shard rather than one unlucky neighbour. The mechanics — computing the affected token ranges, sequencing the moves so no shard is simultaneously donor and recipient, and pacing the pipeline against replication headroom — are covered in depth in the Shard Rebalancing & Data Movement sub-guide.

Two rules from that guide are worth stating here because violating them is so common. First, never change the ring and the data at the same time: the new topology is published in a “pending” state that routing ignores, data moves under the migration pipeline, and only verified keyspaces flip to the new owner. Second, rebalance serially or with a small bounded concurrency (two to three moves in flight), because each move consumes WAL bandwidth, replication slots, and backfill I/O on both its donor and recipient — a fully parallel rebalance is indistinguishable from an incident.

Common Mistakes

FAQ

Related

• Zero-Downtime Resharding Workflows — splitting and merging live shards with logical replication, slot management, and multi-target fan-out
• Online Table Repartitioning — converting monolithic tables to partitioned layouts while they serve traffic
• Shard Rebalancing & Data Movement — sequencing keyspace moves after topology changes and pacing them against replication headroom
• Partition Monitoring & Failover Automation — the alerting stack that watches slot retention, apply lag, and skew before and after every move
• Hash Routing Algorithms — consistent hashing and virtual nodes, the routing substrate that bounds rebalancing cost