DEV Community

Arash Zand
Arash Zand

Posted on • Originally published at Medium

Migrating 200 Million Records in .NET: From Row-by-Row to a Crash-Safe Batched Pipeline

#dotnet #csharp #sql #postgres

A real-world walkthrough of how we rewrote a data migration four times before it was production-ready.

We had to move email history records out of a legacy SQL Server system (16 regional databases, ~200 million rows total) into a PostgreSQL privacy service already running in production. The job had to be idempotent, restartable without duplicates, survive network drops, and finish fast enough to matter.

We went through five versions. Here's what changed each time and why.

The Setup

Source (SQL Server — 16 regional DBs):

CREATE TABLE [dbo].[PlayerEmailHistory] (
    [HistoryId] INT           NOT NULL IDENTITY(1,1) PRIMARY KEY,
    [PlayerId]  INT           NOT NULL,
    [Email]     NVARCHAR(500) NULL,   -- nullable, some accounts wiped
    [ChangedAt] DATETIME      NULL    -- nullable, early records have no timestamp
);

CREATE TABLE [dbo].[PlayerProfileMap] (
    [PlayerId]  INT         NOT NULL PRIMARY KEY,
    [ProfileId] VARCHAR(50) NULL  -- format: "EU|12345"
);
Enter fullscreen mode Exit fullscreen mode

Target (PostgreSQL):

CREATE TABLE "EmailHistory" (
    "Id"        uuid                     NOT NULL PRIMARY KEY,
    "LoginId"   text                     NOT NULL,
    "ProfileId" text                     NOT NULL,
    "Email"     text                     NOT NULL,
    "ChangedBy" text                     NOT NULL,
    "DateUtc"   timestamp with time zone NOT NULL
);
-- Already had a unique index from a previous migration:
CREATE UNIQUE INDEX "UX_EmailHistory_ProfileId_Email_DateUtc"
    ON "EmailHistory" ("ProfileId", "Email", "DateUtc");
Enter fullscreen mode Exit fullscreen mode

Column mapping:

Target Source Resolution
Id — New UUID v7
LoginId PlayerProfileMap.ProfileId JOIN on PlayerId
ProfileId PlayerProfileMap.ProfileId Same as LoginId
Email PlayerEmailHistory.Email Validate
ChangedBy — "legacy-{region}:{HistoryId}"
DateUtc ChangedAt Cast to UTC; UnixEpoch if null

V1: Load Everything, Insert Row by Row

The obvious first attempt. Read all rows into memory, loop, insert each one with WHERE NOT EXISTS.

What goes wrong immediately:

  • Memory — 200M rows in a List<> = >20 GB heap. Process dies with OutOfMemoryException before the first insert.
  • Speed — at 1 ms per row-insert round trip: 200,000 seconds ≈ 7 days.
  • Wrong dedup key — WHERE NOT EXISTS checks (ProfileId, Email, DateUtc). But DateUtc is UnixEpoch for all rows with null ChangedAt. Three rows for the same player with no timestamp → same tuple → two silently dropped.
  • No resume — crash at row 15M, start over from zero.
  • Empty ChangedBy — no way to trace which rows came from which source.

V2: Fix the Dedup Key

Before touching performance, fix correctness.

Key insight: HistoryId is the source PK — unique, stable, never null. Embed it in ChangedBy as "legacy-{region}:{HistoryId}". Now the dedup check becomes:

WHERE NOT EXISTS (SELECT 1 FROM "EmailHistory" WHERE "ChangedBy" = @ChangedBy)
Enter fullscreen mode Exit fullscreen mode

This handles the null-timestamp collision and gives full auditability — any row in the target traces back to an exact source row.

Why the region prefix matters: HistoryId starts at 1 in every regional DB. DK row HistoryId=1 and SE row HistoryId=1 go into the same EU PostgreSQL cluster. Without "legacy-DK:1" vs "legacy-SE:1", one silently disappears.

V2 fixes correctness. Memory and speed are unchanged.

V3: Batched Read + Batch Upsert (~37x faster)

Two changes that together give an order-of-magnitude speedup.

Cursor-based read instead of loading everything:

SELECT TOP (@BatchSize)
    eh.HistoryId, eh.PlayerId, eh.Email, eh.ChangedAt, pm.ProfileId
FROM dbo.PlayerEmailHistory eh WITH (NOLOCK)
LEFT JOIN dbo.PlayerProfileMap pm WITH (NOLOCK)
    ON pm.PlayerId = eh.PlayerId
WHERE eh.HistoryId > @LastHistoryId
ORDER BY eh.HistoryId
Enter fullscreen mode Exit fullscreen mode

Start at @LastHistoryId = 0, advance after each batch. O(1) per batch, no OFFSET penalty.

Batch upsert instead of row-by-row:

private static async Task<ulong> UpsertChunkAsync(
    NpgsqlConnection conn, TargetRecord[] chunk)
{
    var sql        = new StringBuilder();
    var parameters = new DynamicParameters();

    sql.Append("""
        INSERT INTO "EmailHistory"
            ("Id","LoginId","ProfileId","Email","ChangedBy","DateUtc")
        VALUES
        """);

    for (var i = 0; i < chunk.Length; i++)
    {
        if (i > 0) sql.Append(',');
        sql.Append($"(@id{i},@li{i},@pi{i},@em{i},@cb{i},@du{i})");
        parameters.Add($"id{i}", chunk[i].Id);
        parameters.Add($"li{i}", chunk[i].LoginId);
        parameters.Add($"pi{i}", chunk[i].ProfileId);
        parameters.Add($"em{i}", chunk[i].Email);
        parameters.Add($"cb{i}", chunk[i].ChangedBy);
        parameters.Add($"du{i}", chunk[i].DateUtc);
    }
    sql.Append(" ON CONFLICT DO NOTHING;");

    return (ulong)await conn.ExecuteAsync(
        sql.ToString(), parameters, commandTimeout: 600);
}
Enter fullscreen mode Exit fullscreen mode

Why ON CONFLICT DO NOTHING without a target?
The table already has UX_EmailHistory_ProfileId_Email_DateUtc from a previous migration. ON CONFLICT ("ChangedBy") DO NOTHING only handles conflicts on that index — any conflict on the existing index still throws 23505. Omitting the target tells PostgreSQL to silently skip any constraint violation.

Chunk size: 1,000 rows. Very large chunks can exceed SQL string limits; 500–1,500 is a safe range.

V4: Checkpoint (The Production Version)

V3 takes ~4.6 hours per full run. A mid-run crash means starting over. For 16 regions that's painful.

Add a checkpoint table to save the last processed HistoryId after every batch:

CREATE TABLE IF NOT EXISTS "MigrationCheckpointLegacy" (
    "MigrationName" text PRIMARY KEY,
    "LastHistoryId" int  NOT NULL,
    "UpdatedAtUtc"  timestamp with time zone NOT NULL DEFAULT NOW()
);
Enter fullscreen mode Exit fullscreen mode

Name it specifically. The privacy service already had EmailMigrationCheckpoint from a MySQL migration with different columns. CREATE TABLE IF NOT EXISTS silently succeeds — the table exists — then your column query fails with 42703: column "LastHistoryId" does not exist. Naming it MigrationCheckpointLegacy avoids the collision.

Add a partial unique index instead of a full one:

CREATE UNIQUE INDEX IF NOT EXISTS "UX_EmailHistory_ChangedBy_Legacy"
ON "EmailHistory" ("ChangedBy")
WHERE "ChangedBy" LIKE 'legacy-%';
Enter fullscreen mode Exit fullscreen mode

A full unique index on ChangedBy fails if any two existing rows (from the previous migration) share the same value. A partial index scoped to our own rows always succeeds.

Checkpoint save — upsert after every batch:

await conn.ExecuteAsync("""
    INSERT INTO "MigrationCheckpointLegacy"
        ("MigrationName", "LastHistoryId", "UpdatedAtUtc")
    VALUES (@MigrationName, @LastHistoryId, NOW())
    ON CONFLICT ("MigrationName")
    DO UPDATE SET
        "LastHistoryId" = EXCLUDED."LastHistoryId",
        "UpdatedAtUtc"  = EXCLUDED."UpdatedAtUtc"
    """, new { MigrationName = migrationKey, LastHistoryId = lastHistoryId });
Enter fullscreen mode Exit fullscreen mode

The --region flag — run one region at a time:

// Program.cs
var regionFilter = GetArg(args, "--region");
if (regionFilter is not null)
    options.Regions = [options.Regions.First(
        r => r.Name.Equals(regionFilter, StringComparison.OrdinalIgnoreCase))];
Enter fullscreen mode Exit fullscreen mode 
dotnet run -- --region DK
dotnet run -- --region SE
Enter fullscreen mode Exit fullscreen mode

V5: Channel-Based Producer-Consumer Pipeline

V4 is sequential: read batch N → write batch N → read batch N+1. Both I/O operations never overlap.

System.Threading.Channels (built into .NET, no NuGet required) lets us pipeline them:

V4: [read N]──[write N]──[read N+1]──[write N+1]──...
V5: [read N]──[read N+1]──[read N+2]──...
         [write N]──[write N+1]──[write N+2]──...
Enter fullscreen mode Exit fullscreen mode

Channel setup — bounded to prevent lookahead memory explosion:

private record BatchPayload(List<TargetRecord> Records, int LastHistoryId);

private static Channel<BatchPayload> CreateChannel() =>
    Channel.CreateBounded<BatchPayload>(
        new BoundedChannelOptions(capacity: 3)
        {
            SingleWriter = true,
            SingleReader = true,
            FullMode = BoundedChannelFullMode.Wait
        });
Enter fullscreen mode Exit fullscreen mode

Capacity 3 = 3 pre-read batches maximum. Without the bound, the producer runs arbitrarily far ahead — same memory problem as V1.

Producer — finally block is critical:

private async Task ProduceAsync(
    RegionConfig region, int startCheckpoint,
    ChannelWriter<BatchPayload> writer, Action<ulong> onSkipped)
{
    var checkpoint = startCheckpoint;
    try
    {
        while (true)
        {
            var batch = await ReadSourceBatchAsync(region, checkpoint);
            if (batch.Count == 0) break;

            var (valid, skipped) = FilterAndTransform(batch, region.Name);
            onSkipped(skipped);
            checkpoint = batch[^1].HistoryId;

            await writer.WriteAsync(new BatchPayload(valid, checkpoint));
        }
    }
    finally
    {
        writer.Complete(); // critical: consumer must exit even if producer throws
    }
}
Enter fullscreen mode Exit fullscreen mode

Consumer:

private async Task<(ulong inserted, ulong duplicates)> ConsumeAsync(
    NpgsqlConnection conn, string migrationKey,
    ChannelReader<BatchPayload> reader)
{
    var inserted = 0UL; var duplicates = 0UL;

    await foreach (var payload in reader.ReadAllAsync())
    {
        var n = await UpsertBatchAsync(conn, payload.Records);
        inserted   += n;
        duplicates += (ulong)payload.Records.Count - n;
        await SaveCheckpointAsync(conn, migrationKey, payload.LastHistoryId);
    }
    return (inserted, duplicates);
}
Enter fullscreen mode Exit fullscreen mode

Wiring:

var channel      = CreateChannel();
var producerTask = ProduceAsync(region, checkpoint, channel.Writer,
                                count => Interlocked.Add(ref skipped, count));
var consumerTask = ConsumeAsync(conn, migrationKey, channel.Reader);

await Task.WhenAll(producerTask, consumerTask);
Enter fullscreen mode Exit fullscreen mode

Interlocked.Add on the skipped counter — producer and consumer run concurrently, shared state needs atomic updates.

Benchmark Comparison

Version Peak Memory Insert Rate Est. Time (200M) Crash Recovery
V1: row-by-row, all in memory >20 GB ~320 rows/sec ~7 days Start over
V2: row-by-row, fixed dedup >20 GB ~290 rows/sec ~8 days Start over
V3: batch read + batch upsert ~50 MB ~12,000 rows/sec ~4.6 hours Start over
V4: batched + checkpoint ~50 MB ~11,500 rows/sec ~4.8 hours Resume from last batch
V5: channel pipeline + checkpoint ~55 MB ~17,000 rows/sec ~3.3 hours Resume from last batch

Our production run across all 16 regions took ~47 hours total. The checkpoint system recovered from two mid-run network failures without data loss or manual intervention.

Production War Stories

1. SSL Certificate Not Trusted
Microsoft.Data.SqlClient v4+ enforces encrypted connections and validates server certs. RDS self-signed cert → connection fails. Fix: TrustServerCertificate=True in the SQL Server connection string (fine for an internal migration script, not for application code).

2. Checkpoint Table Collision
The privacy service already had a migration checkpoint table with different columns. CREATE TABLE IF NOT EXISTS silently succeeded, then our column query failed with 42703. Fix: name your checkpoint table specifically for your migration (MigrationCheckpointLegacy).

3. Pre-Existing Unique Constraint
ON CONFLICT ("ChangedBy") DO NOTHING only handles conflicts on that index. A conflict on the existing UX_EmailHistory_ProfileId_Email_DateUtc still throws 23505. Fix: use ON CONFLICT DO NOTHING without specifying a target.

4. Cross-Region HistoryId Collision
HistoryId starts at 1 in every regional DB. Two regions, same HistoryId, same cluster → one row silently lost. Fix: include region in ChangedBy ("legacy-DK:1" vs "legacy-SE:1").

5. Error 4060 from SQL Server
The server connection succeeded but the login doesn't have access to that specific database. Not a network issue — a GRANT issue. Update credentials per region.

6. Network Drop Mid-Migration
System.IO.IOException: An existing connection was forcibly closed by the remote host. Without checkpoint: restart from scratch. With checkpoint: resume from last saved batch, ~1,000 rows re-processed as duplicates (silently skipped), re-run completed in under a minute.

Key Takeaways

Performance: Row-by-row insert doesn't scale. Batch upsert with ON CONFLICT DO NOTHING is an order of magnitude faster. Read with a cursor (WHERE id > @lastId), not OFFSET.

Resilience: Save a checkpoint after every batch. Name checkpoint tables specifically so they don't collide with other migrations. Use partial unique indexes when sharing a table with other writers.

Pipelining: If read and write are both I/O-bound, System.Threading.Channels lets them overlap. Bound the channel. Call writer.Complete() in a finally block. Embed the checkpoint value in the batch payload to avoid shared mutable state.

Operations: Run regions one at a time with a --region flag. Add a --progress command. Set explicit command timeouts (commandTimeout: 600) — Dapper's default 30s isn't enough for batch inserts into busy tables.

Originally published on Medium. The full article includes complete V3 and V4 migrator code, configuration models, and the --progress CLI command.

Top comments (0)