Data Migration

Is downtime acceptable?
  YES -> How long?
    < 1 hour -> Simple dump & restore with maintenance window
    < 15 min -> Blue-green with quick switchover
    Any length -> Stop-the-world migration
  NO -> Zero-downtime migration required
    What is changing?
      Schema only -> Online schema migration (pt-osc, gh-ost, pg_repack)
      Database engine -> Dual-write + shadow read migration
      Data model -> Expand-contract pattern
      Cloud provider -> Replicate + cutover migration

Phase 1 - EXPAND: Add new structure alongside old
Phase 2 - MIGRATE: Backfill data, dual-write
Phase 3 - SWITCH: Point reads to new structure
Phase 4 - CONTRACT: Remove old structure

Timeline:
  Deploy 1 (Expand)    Deploy 2 (Switch)    Deploy 3 (Contract)
  |------ dual write ------|---- reads from new ----|
  |-- backfill old data ---|                        |
                            |-- verify correctness --|
                                                     |-- drop old --|

-- Goal: Split "full_name" into "first_name" and "last_name"

-- Phase 1: EXPAND - Add new columns (non-breaking)
ALTER TABLE customers ADD COLUMN first_name VARCHAR(100);
ALTER TABLE customers ADD COLUMN last_name VARCHAR(100);

-- Phase 2: MIGRATE - Deploy code that writes to both old and new
-- Application code:
-- def save_customer(customer):
--     customer.full_name = f"{customer.first_name} {customer.last_name}"
--     # Writes to all three columns
--     db.save(customer)

-- Backfill existing data
# ... (condensed) ...
WHERE first_name IS NULL OR last_name IS NULL;
-- Should return 0

-- Phase 4: CONTRACT - Remove old column (after verification period)
ALTER TABLE customers DROP COLUMN full_name;

-- Goal: Change "price" from INTEGER (cents) to NUMERIC(10,2) (dollars)

-- Phase 1: EXPAND
ALTER TABLE products ADD COLUMN price_decimal NUMERIC(10,2);

-- Phase 2: BACKFILL + DUAL-WRITE
UPDATE products SET price_decimal = price / 100.0 WHERE price_decimal IS NULL;

-- Deploy: application writes to both columns
-- CREATE TRIGGER for safety:
CREATE OR REPLACE FUNCTION sync_price() RETURNS TRIGGER AS $$
BEGIN
    IF NEW.price IS DISTINCT FROM OLD.price THEN
        NEW.price_decimal := NEW.price / 100.0;
    # ... (condensed) ...
-- Phase 3: SWITCH reads to price_decimal
-- Phase 4: Drop trigger, rename column, drop old
DROP TRIGGER price_sync ON products;
ALTER TABLE products DROP COLUMN price;
ALTER TABLE products RENAME COLUMN price_decimal TO price;

class DualWriteService:
    """Write to both old and new database during migration."""

    def __init__(self, old_db, new_db, mode='shadow'):
        self.old_db = old_db
        self.new_db = new_db
        self.mode = mode  # shadow, dual, cutover

    def write(self, entity):
        if self.mode == 'shadow':
            # Old DB is primary, async write to new DB
            result = self.old_db.save(entity)
            try:
                self.new_db.save(entity)
            # ... (condensed) ...
                              f"old={primary_result}, new={secondary}")
            except Exception as e:
                metrics.increment("shadow_read.errors")

        threading.Thread(target=compare, daemon=True).start()

Week 1-2: shadow mode
  - Write old (primary) + new (async)
  - Read old only
  - Monitor mismatch rate

Week 3: dual mode
  - Write both (synchronous)
  - Read old
  - Mismatch rate should be ~0%

Week 4: cutover mode
  - Write new (primary) + old (async)
  - Read new
  - Monitor for issues

Week 5+: decommission
  - Remove old DB writes
  - Archive old DB

import hashlib
from typing import Dict, List, Tuple

class DataReconciler:
    """Compare data between source and target databases."""

    def __init__(self, source_engine, target_engine):
        self.source = source_engine
        self.target = target_engine

    def reconcile_counts(self, table_pairs: List[Tuple[str, str]]) -> List[dict]:
        """Compare row counts between source and target tables.

        Note: Table/column names are interpolated via f-strings because SQL
        # ... (condensed) ...
                'target': float(t_val) if t_val else None,
                'pct_difference': round(pct_diff, 4),
                'match': pct_diff < 0.01,  # Within 0.01%
            }
        return results

BACKWARD COMPATIBLE (safe to deploy new code first):
  + Add nullable column
  + Add column with default value
  + Widen column type (INT -> BIGINT, VARCHAR(50) -> VARCHAR(100))
  + Add new table
  + Add index

NOT BACKWARD COMPATIBLE (requires expand-contract):
  - Remove column
  - Rename column
  - Narrow column type
  - Add NOT NULL constraint to existing column
  - Change column type (INT -> VARCHAR)

FORWARD COMPATIBLE (safe to deploy old code with new schema):
  + Remove column (if old code does not reference it)
  + Add nullable column (old code ignores it)

migrations/
  V001__create_users_table.sql
  V002__add_email_to_users.sql
  V003__create_orders_table.sql
  V004__add_index_orders_user_id.sql
  V005__split_name_columns_expand.sql      # Phase 1: Add columns
  V006__split_name_columns_backfill.sql    # Phase 2: Populate data
  V007__split_name_columns_contract.sql    # Phase 4: Remove old column

-- V005__split_name_columns_expand.sql
-- Expand phase: add new columns

ALTER TABLE customers ADD COLUMN first_name VARCHAR(100);
ALTER TABLE customers ADD COLUMN last_name VARCHAR(100);

-- Add trigger for dual-write during transition
CREATE OR REPLACE FUNCTION sync_name_columns()
RETURNS TRIGGER AS $$
BEGIN
    IF NEW.full_name IS NOT NULL AND (NEW.first_name IS NULL OR TG_OP = 'INSERT') THEN
        NEW.first_name := SPLIT_PART(NEW.full_name, ' ', 1);
        NEW.last_name := SUBSTRING(NEW.full_name FROM POSITION(' ' IN NEW.full_name) + 1);
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_sync_names
BEFORE INSERT OR UPDATE ON customers
FOR EACH ROW EXECUTE FUNCTION sync_name_columns();

-- Every migration should have a corresponding rollback script
-- V005__split_name_columns_expand.sql -> V005__rollback.sql

-- V005__rollback.sql
DROP TRIGGER IF EXISTS trg_sync_names ON customers;
DROP FUNCTION IF EXISTS sync_name_columns();
ALTER TABLE customers DROP COLUMN IF EXISTS first_name;
ALTER TABLE customers DROP COLUMN IF EXISTS last_name;

class MigrationWithRollback:
    """Migration with automatic rollback on failure."""

    def __init__(self, source_engine, target_engine):
        self.source = source_engine
        self.target = target_engine
        self.rollback_log = []

    def migrate_with_snapshot(self, table, migration_fn):
        """Take snapshot before migration for safe rollback.

        Note: Table names are interpolated via f-strings below because SQL
        parameters cannot be used for identifiers. Validate that table names
        are never derived from user input.
        # ... (condensed) ...
    def rollback(self, table, snapshot_table):
        """Restore table from snapshot."""
        with self.target.begin() as conn:
            conn.execute(f"DROP TABLE IF EXISTS {table}")
            conn.execute(f"ALTER TABLE {snapshot_table} RENAME TO {table}")

# gh-ost (GitHub Online Schema Tester) for MySQL
gh-ost \
    --host=primary-db \
    --database=production \
    --table=orders \
    --alter="ADD COLUMN shipping_status VARCHAR(20) DEFAULT 'pending'" \
    --allow-on-master \
    --chunk-size=1000 \
    --max-load="Threads_running=25" \
    --critical-load="Threads_running=50" \
    --exact-rowcount \
    --concurrent-rowcount \
    --default-retries=120 \
    --postpone-cut-over-flag-file=/tmp/gh-ost.postpone \
    --execute

# How gh-ost works:
# 1. Creates ghost table with new schema
# 2. Copies rows from original to ghost in chunks
# 3. Applies binary log events for ongoing changes
# 4. Atomically swaps tables when caught up

-- PostgreSQL: pg_repack for table reorganization
-- Does not require ACCESS EXCLUSIVE lock for the full duration

-- Install extension
CREATE EXTENSION pg_repack;

-- Repack a table (remove bloat, apply new storage parameters)
-- Run from command line:
-- pg_repack --table orders --no-superuser-check -d production

def migrate_large_table(source_engine, target_engine, table,
                        key_column, chunk_size=10000):
    """
    Migrate large table in chunks with progress tracking and resumability.

    Note: Table/column names are interpolated via f-strings because SQL
    parameters cannot be used for identifiers. Ensure these values are
    never derived from user input.
    """
    # Get total count and max key for progress tracking
    total = pd.read_sql(f"SELECT COUNT(*) AS c FROM {table}", source_engine).iloc[0]['c']

    # Get resume point
    last_key = get_checkpoint(table)
    # ... (condensed) ...
        migrated += len(chunk)
        progress = migrated / total * 100
        print(f"Progress: {migrated}/{total} ({progress:.1f}%)")

    return migrated

# MySQL to PostgreSQL type mapping
MYSQL_TO_PG_TYPES = {
    'TINYINT':    'SMALLINT',
    'SMALLINT':   'SMALLINT',
    'MEDIUMINT':  'INTEGER',
    'INT':        'INTEGER',
    'BIGINT':     'BIGINT',
    'FLOAT':      'REAL',
    'DOUBLE':     'DOUBLE PRECISION',
    'DECIMAL':    'NUMERIC',
    'CHAR':       'CHAR',
    'VARCHAR':    'VARCHAR',
    'TINYTEXT':   'TEXT',
    'TEXT':       'TEXT',
    # ... (condensed) ...
    'UNIQUEIDENTIFIER': 'UUID',
    'VARBINARY':      'BYTEA',
    'IMAGE':          'BYTEA',
    'XML':            'XML',
}