Designing a Data Migration Pipeline: Schema Mapping, Incremental Sync, and Zero-Downtime Cutover

Most migrations fail in the details. Not in the bulk copy, but in the ten days before cutover when you discover that the source system has 40,000 rows that violate a constraint you assumed was enforced, that two columns have swapped semantic meaning over the past three years, and that the legacy platform batches deletes at midnight in a way that your CDC stream does not capture.

A data migration pipeline is not a script. It is a set of phased systems: a schema-mapping layer that translates between data models, a sync mechanism that keeps the target current without hammering the source, a dual-write period that maintains consistency while you validate, and a cutover sequence with a defined rollback path. Each phase has its own failure modes.

This article walks through each layer with concrete code, tradeoffs, and the production details that are usually omitted from migration guides.

Framing the Problem

There are two broad migration scenarios that share most of the same architecture:

Legacy modernization (PE portfolio companies, enterprise rewrites): you are moving from a monolith or on-premise database to a new schema, often across vendors (Oracle to PostgreSQL, SQL Server to Aurora). The source is live and cannot be taken down. The cutover window is measured in minutes, not hours.

Platform escape (startups leaving no-code tools): you are moving from Airtable, Bubble, or a Firebase document model to a relational or purpose-built database. The data model is usually denormalized, relationships are implicit, and historical data quality is inconsistent.

Both benefit from the same pipeline structure. The schema-mapping complexity differs, but the sync and cutover mechanisms are identical.

Phase 1: Schema Mapping and Transformation

Schema mapping is the part where you decide what the source data means and what shape it needs to take in the target. Do this before writing any sync code.

Data Profiling First

Before you can map schemas, you need to know what is actually in the source database, not just what the schema says. Column types lie. A VARCHAR(255) might contain JSON, pipe-delimited values, or free-text email addresses that are not valid emails. An INT foreign key might reference rows that no longer exist.

Run a profiling pass:

interface ColumnProfile {
  table: string;
  column: string;
  dataType: string;
  nullCount: number;
  distinctCount: number;
  minLength: number;
  maxLength: number;
  sampleValues: unknown[];
  violatesNotNull: boolean;
  violatesForeignKey: boolean;
}

async function profileColumn(
  db: DatabaseClient,
  table: string,
  column: string,
  totalRows: number
): Promise<ColumnProfile> {
  const [stats] = await db.query<{
    null_count: number;
    distinct_count: number;
    min_len: number;
    max_len: number;
  }>(`
    SELECT
      COUNT(*) FILTER (WHERE ${column} IS NULL) AS null_count,
      COUNT(DISTINCT ${column})                  AS distinct_count,
      MIN(LENGTH(${column}::text))               AS min_len,
      MAX(LENGTH(${column}::text))               AS max_len
    FROM ${table}
  `);

  const samples = await db.query<Record<string, unknown>>(
    `SELECT ${column} FROM ${table} WHERE ${column} IS NOT NULL LIMIT 10`
  );

  return {
    table,
    column,
    dataType: "VARCHAR", // resolved from information_schema separately
    nullCount: stats.null_count,
    distinctCount: stats.distinct_count,
    minLength: stats.min_len,
    maxLength: stats.max_len,
    sampleValues: samples.map((r) => r[column]),
    violatesNotNull: stats.null_count > 0,
    violatesForeignKey: false, // filled by a separate FK check pass
  };
}

Profile every column. Pay attention to maxLength versus the declared column size, null rates on columns the application assumes are non-null, and distinct counts that reveal whether a column is used as an enum despite not having a constraint.

Mapping Rules as Data

Represent your mapping rules as configuration, not imperative code. This lets you validate, audit, and version the mapping separately from the execution engine.

type TransformFn = (value: unknown, row: Record<string, unknown>) => unknown;

interface ColumnMapping {
  sourceColumn: string;
  targetColumn: string;
  transform?: TransformFn;
  defaultValue?: unknown;
  required: boolean;
}

interface TableMapping {
  sourceTable: string;
  targetTable: string;
  columns: ColumnMapping[];
  filter?: string; // SQL WHERE clause to exclude rows
  batchSize: number;
}

const orderMapping: TableMapping = {
  sourceTable: "legacy_orders",
  targetTable: "orders",
  batchSize: 1000,
  filter: "status != 'DELETED'",
  columns: [
    {
      sourceColumn: "order_id",
      targetColumn: "id",
      required: true,
    },
    {
      sourceColumn: "cust_id",
      targetColumn: "customer_id",
      required: true,
    },
    {
      sourceColumn: "total",
      targetColumn: "total_cents",
      // Legacy stores dollars as DECIMAL(10,2), target stores cents as INT
      transform: (value) => Math.round(Number(value) * 100),
      required: true,
    },
    {
      sourceColumn: "order_status",
      targetColumn: "status",
      transform: (value) => STATUS_MAP[value as string] ?? "unknown",
      required: true,
    },
    {
      sourceColumn: "created",
      targetColumn: "created_at",
      transform: (value) => new Date(value as string).toISOString(),
      required: true,
    },
  ],
};

const STATUS_MAP: Record<string, string> = {
  "0": "pending",
  "1": "confirmed",
  "2": "shipped",
  "3": "delivered",
  "4": "cancelled",
  P: "pending",
  C: "confirmed",
  S: "shipped",
};

The transform function receives the full source row, not just the value, so you can derive target columns from multiple source columns. Log every row where a transform produces a null for a required column: these are your data quality failures.

Phase 2: Initial Bulk Load

Once you have validated your mapping against a sample, run the initial bulk load against a point-in-time snapshot. Do not try to sync against a live database during the bulk load phase.

Batched Copy with Progress Tracking

interface MigrationProgress {
  tableMapping: string;
  totalRows: number;
  processedRows: number;
  failedRows: number;
  startedAt: Date;
  checkpointOffset: number;
}

async function bulkLoad(
  mapping: TableMapping,
  sourceDb: DatabaseClient,
  targetDb: DatabaseClient,
  progressStore: ProgressStore
): Promise<void> {
  const totalRows = await countRows(sourceDb, mapping.sourceTable, mapping.filter);
  let offset = await progressStore.getCheckpoint(mapping.sourceTable) ?? 0;

  console.log(`Bulk loading ${mapping.sourceTable}: ${totalRows} rows, resuming from offset ${offset}`);

  while (offset < totalRows) {
    const sourceRows = await sourceDb.query<Record<string, unknown>>(`
      SELECT *
      FROM ${mapping.sourceTable}
      ${mapping.filter ? `WHERE ${mapping.filter}` : ""}
      ORDER BY rowid
      LIMIT ${mapping.batchSize}
      OFFSET ${offset}
    `);

    if (sourceRows.length === 0) break;

    const { transformed, failed } = applyMapping(mapping, sourceRows);

    if (transformed.length > 0) {
      await targetDb.batchInsert(mapping.targetTable, transformed, {
        onConflict: "update", // idempotent: safe to re-run from checkpoint
      });
    }

    if (failed.length > 0) {
      await progressStore.logFailures(mapping.sourceTable, offset, failed);
    }

    offset += sourceRows.length;
    await progressStore.saveCheckpoint(mapping.sourceTable, offset);

    if (offset % 50_000 === 0) {
      console.log(`  ${offset}/${totalRows} rows processed`);
    }
  }
}

Key decisions here: order by a stable column (primary key or rowid), checkpoint after every batch so the load is resumable, and use upsert semantics so that rerunning after a failure does not create duplicates.

For tables over 50 million rows, parallelize by range-partitioning the primary key across workers. Use a coordinator that divides the key space into N equal ranges and dispatches each range to a worker process.

Phase 3: CDC-Based Incremental Sync

After the bulk load, the target database has a consistent snapshot as of some point in time. The source has continued accepting writes. You need to apply every change that occurred during and after the bulk load without re-running the full copy.

This is where change data capture (CDC) enters. The bulk load note a start LSN (PostgreSQL) or binlog position (MySQL) before it begins. After the bulk load completes, the CDC consumer replays all changes from that start position through the present, then continues streaming in real time.

CDC Consumer with Mapping Applied

import { LogicalReplicationService, PgoutputPlugin } from "pg-logical-replication";

interface CdcEvent {
  op: "insert" | "update" | "delete";
  table: string;
  before: Record<string, unknown> | null;
  after: Record<string, unknown> | null;
  lsn: string;
}

async function startIncrementalSync(
  mappings: Map<string, TableMapping>,
  sourceConfig: { connectionString: string; slotName: string; publicationName: string },
  targetDb: DatabaseClient,
  startLsn: string
): Promise<void> {
  const service = new LogicalReplicationService({
    connectionString: sourceConfig.connectionString,
  });

  const plugin = new PgoutputPlugin({
    protoVersion: 1,
    publicationNames: [sourceConfig.publicationName],
  });

  service.on("data", async (lsn: string, rawEvent: unknown) => {
    const event = parseWalEvent(rawEvent);
    if (!event) {
      await service.acknowledge(lsn);
      return;
    }

    const mapping = mappings.get(event.table);
    if (!mapping) {
      await service.acknowledge(lsn);
      return;
    }

    await applyIncrementalEvent(event, mapping, targetDb);
    await service.acknowledge(lsn);
  });

  await service.subscribe(plugin, sourceConfig.slotName, startLsn);
}

async function applyIncrementalEvent(
  event: CdcEvent,
  mapping: TableMapping,
  targetDb: DatabaseClient
): Promise<void> {
  switch (event.op) {
    case "insert":
    case "update": {
      if (!event.after) return;
      const { transformed } = applyMapping(mapping, [event.after]);
      if (transformed.length > 0) {
        await targetDb.upsert(mapping.targetTable, transformed[0]);
      }
      break;
    }
    case "delete": {
      if (!event.before) return;
      const sourceKey = event.before[getPrimaryKeyColumn(mapping)];
      const targetKey = mapPrimaryKey(mapping, sourceKey);
      await targetDb.delete(mapping.targetTable, { id: targetKey });
      break;
    }
  }
}

The startLsn parameter is the position recorded before the bulk load began. During the initial catch-up phase, events will overlap with rows already inserted by the bulk load. The upsert semantics on insert and update handle this correctly. For deletes during the overlap window, you may delete rows that were never inserted to the target yet: the target delete is a no-op, which is correct.

Phase 4: Dual-Write

When incremental sync has caught up to within a few seconds of the source, you enter the dual-write phase. Application writes now go to both the source and target simultaneously. This serves two purposes: it validates that your write path produces correct data in the target schema, and it provides a fast rollback path during cutover (stop writing to the new system, revert reads).

class DualWriteClient {
  constructor(
    private readonly primary: DatabaseClient,   // source (legacy)
    private readonly secondary: DatabaseClient, // target (new)
    private readonly mode: "primary-only" | "dual-write" | "secondary-only"
  ) {}

  async upsertOrder(order: OrderInput): Promise<void> {
    if (this.mode === "secondary-only") {
      await this.secondary.upsert("orders", toNewSchema(order));
      return;
    }

    // Always write to primary first
    await this.primary.upsert("legacy_orders", toLegacySchema(order));

    if (this.mode === "dual-write") {
      try {
        await this.secondary.upsert("orders", toNewSchema(order));
      } catch (err) {
        // Secondary write failure does not block the primary write.
        // Log and alert, but do not throw.
        metrics.increment("dual_write.secondary_failure", { table: "orders" });
        logger.error("Dual-write secondary failure", { order, err });
      }
    }
  }
}

Secondary write failures during dual-write must not block the primary write. The source system is the source of truth until cutover. If the secondary falls behind, your CDC sync will catch it up. The dual-write layer validates the write path; the CDC sync is the safety net.

Phase 5: Validation and Reconciliation

Before cutover, you need automated proof that the target contains the right data. Manual spot-checking does not scale, and it misses the edge cases that matter.

Row-Count and Checksum Reconciliation

interface ReconciliationResult {
  table: string;
  sourceCount: number;
  targetCount: number;
  countMatch: boolean;
  sampleMismatchCount: number;
  checksumMatch: boolean;
}

async function reconcileTable(
  mapping: TableMapping,
  sourceDb: DatabaseClient,
  targetDb: DatabaseClient,
  sampleSize: number = 10_000
): Promise<ReconciliationResult> {
  const [sourceCount, targetCount] = await Promise.all([
    countRows(sourceDb, mapping.sourceTable, mapping.filter),
    countRows(targetDb, mapping.targetTable),
  ]);

  // Checksum a random sample of rows
  const sampleKeys = await sourceDb.query<{ id: unknown }>(`
    SELECT id FROM ${mapping.sourceTable}
    ${mapping.filter ? `WHERE ${mapping.filter}` : ""}
    ORDER BY RANDOM()
    LIMIT ${sampleSize}
  `);

  let mismatchCount = 0;

  for (const { id } of sampleKeys) {
    const [sourceRow] = await sourceDb.query(
      `SELECT * FROM ${mapping.sourceTable} WHERE id = $1`,
      [id]
    );
    const [targetRow] = await targetDb.query(
      `SELECT * FROM ${mapping.targetTable} WHERE id = $1`,
      [id]
    );

    if (!targetRow) {
      mismatchCount++;
      continue;
    }

    const sourceTransformed = applyMapping(mapping, [sourceRow]).transformed[0];
    if (!deepEqual(sourceTransformed, targetRow)) {
      mismatchCount++;
    }
  }

  return {
    table: mapping.sourceTable,
    sourceCount,
    targetCount,
    countMatch: sourceCount === targetCount,
    sampleMismatchCount: mismatchCount,
    checksumMatch: mismatchCount === 0,
  };
}

Run reconciliation continuously during the dual-write phase, not just before cutover. Discrepancies that appear and disappear indicate ordering issues in the sync; discrepancies that persist indicate mapping bugs or missed write paths.

Phase 6: Zero-Downtime Cutover

The cutover sequence transitions reads from source to target. The goal is to make this reversible at every step.

1. Verify sync lag < 1 second
2. Enable dual-write (if not already active)
3. Run final reconciliation pass — abort if mismatch > threshold
4. Pause writes briefly (at the load balancer or application level)
5. Confirm CDC has processed all remaining events (lag = 0)
6. Switch read traffic to target (feature flag or config change)
7. Resume writes, now targeting the new system
8. Monitor error rates for 30 minutes
9. If stable: disable CDC sync, decommission source writes
10. If not stable: revert read traffic to source, re-enable source writes

The pause in step 4 is the only moment of reduced write availability. For most systems this is under 5 seconds. If your application cannot tolerate even a 5-second write pause, you need a more complex cutover using version vectors to detect and reject stale reads rather than a hard pause.

The feature flag switch in step 6 controls read routing:

async function getOrder(orderId: string): Promise<Order> {
  const useNewSystem = await featureFlags.isEnabled("read-from-new-db", {
    userId: "system",
  });

  if (useNewSystem) {
    return newDb.findOrder(orderId);
  } else {
    return legacyDb.findOrder(orderId);
  }
}

Keep both read paths functional for at least 48 hours after cutover. Rollback is a flag flip, not a re-migration.

Tradeoffs

Dimension	CDC-based incremental sync	Scheduled batch sync	Application-level dual-write only
Cutover window	Seconds (lag-driven)	Hours (full re-copy)	Seconds
Source DB load	Low (WAL reading)	High (full scans)	None additional
Delete handling	Native (WAL captures deletes)	Requires soft-delete convention	Native
Complexity	High (CDC infra required)	Low	Medium
Rollback speed	Immediate (flag flip)	Hours	Immediate
Data validation	Continuous reconciliation	Pre-cutover only	Continuous
Suitable for large datasets	Yes	No (at scale)	Yes
Schema divergence tolerance	Medium (transform layer)	Medium	Low (both schemas must coexist)

Batch sync is viable only when the source can be quiesced or you are willing to accept a longer cutover window. For live production systems, CDC is the right foundation.

Production Considerations

Replication slot retention. A CDC replication slot on PostgreSQL retains WAL until the consumer acknowledges. If your sync falls significantly behind, this causes disk pressure on the source. Set max_replication_slots conservatively and alert on slot lag before it becomes a disk crisis.

Schema migrations during the migration window. If developers continue shipping schema changes to the source during the migration period, your mapping layer breaks. Freeze schema changes on the source from the start of the CDC sync phase through cutover. Enforce this with a code review gate, not a policy document.

Identity and foreign key mapping. When the source uses auto-increment integers and the target uses UUIDs, you need a mapping table that translates source IDs to target IDs for every foreign key reference. Build this mapping incrementally during the bulk load and keep it in memory for CDC event processing. At 50 million rows, a full in-memory map is around 800MB (assuming 8 bytes per integer ID + 16 bytes per UUID). If that is too large, use a fast key-value store (Redis) with the mapping as the value.

Partial failures in dual-write. A network partition between your application and the secondary database during dual-write will cause divergence. Your CDC sync will resolve this, but the reconciliation job needs to detect the gap and alert. Do not assume the sync will silently catch up: verify it did.

Timezone and encoding edge cases. Legacy databases often have inconsistent timezone handling, mixed character encodings, or NULL bytes in text fields. These surface only in the data profiling phase and require explicit handling in your transform layer. A NULL byte in a PostgreSQL string causes the insert to fail with a non-obvious error. Strip them in the transform or the target insert will fail silently if you are using a bulk loader that swallows errors.

Testing the rollback path. Teams rehearse the cutover sequence but rarely rehearse the rollback. Before the real cutover, run a full drill: cut over, wait 15 minutes, roll back, verify the source is consistent. You will find things that the forward path rehearsal missed.

A migration pipeline built around these phases gives you something more valuable than a migrated database: a reversible transition with continuous validation and a defined abort path at every step. The migration is not done when the bulk load completes. It is done when you have decommissioned the source and confirmed the target has been stable for long enough that rollback is no longer a realistic option.