PostgreSQL UPSERT with History: From 90 Seconds to 5 with CTE

Hi, a couple of years ago I wrote about batch insert and update in PostgreSQL using UNNEST - it's worth expanding on this topic by using a CTE, based on a real-world task. 100k emails with an audit log were being inserted in 90 seconds. I rewrote it using a CTE - it became 5.

The task: take a file with active email addresses and update their state in the database while preserving the change history. All addresses from the file are marked as active; all addresses that were in the database but are missing from the file are deactivated. Additionally, a change history is recorded for audit purposes - who updated the states and when. In theory, the file could contain anywhere from 100 to 100,000 addresses.

Database Schema

Here's what the database schema looks like for storing state and change history:

CREATE TABLE rtt_email_files (
    id BIGSERIAL NOT NULL PRIMARY KEY,
    -- meta: file name, size, ...
    created_at TIMESTAMP NOT NULL,
    created_by BIGINT NOT NULL REFERENCES users (id)
);

CREATE TABLE rtt_emails (
    id BIGSERIAL NOT NULL PRIMARY KEY,
    email VARCHAR NOT NULL UNIQUE, -- lowercase email address
    active BOOLEAN NOT NULL,
    created_at TIMESTAMP NOT NULL,
    updated_at TIMESTAMP NOT NULL
);

CREATE TABLE rtt_email_history (
    id BIGSERIAL NOT NULL PRIMARY KEY,
    email_id BIGINT NOT NULL REFERENCES rtt_emails (id),
    active BOOLEAN NOT NULL,
    file_id BIGINT NOT NULL REFERENCES rtt_email_files (id),
    created_at TIMESTAMP NOT NULL
);

Original Approach (Multiple Queries)

I use sqlc to interact with PostgreSQL, so here are the queries I prepared:

-- name: RttEmailFilesNew :one
INSERT INTO rtt_email_files (created_at, created_by)
VALUES (@created_at, @created_by)
RETURNING id;

-- name: RttEmailsActivate :many
INSERT INTO rtt_emails AS t (email, active, created_at, updated_at)
VALUES (UNNEST(@emails::VARCHAR[]), TRUE, @created_at, @created_at)
ON CONFLICT (email) DO UPDATE
SET active = EXCLUDED.active, updated_at = EXCLUDED.updated_at
WHERE t.active IS DISTINCT FROM EXCLUDED.active
RETURNING id;

-- name: RttEmailsDeactivate :many
UPDATE rtt_emails
SET active = FALSE, updated_at = @updated_at
WHERE (@emails::VARCHAR[] IS NULL OR NOT email = ANY (@emails::VARCHAR[]))
  AND active = TRUE
RETURNING id;

-- name: RttEmailHistoryNew :exec
INSERT INTO rtt_email_history (email_id, active, file_id, created_at)
VALUES (UNNEST(@email_ids::BIGINT[]), @active, @file_id, @created_at);

And here's what the insertion code looks like:

func (r *RttEmailRepository) Upload(
    ctx context.Context,
    file *FileMetadata,
    emails []string,
    createdAt time.Time,
    createdBy int64,
) (activatedCount int64, deactivatedCount int64, err error) {
    txErr := r.db.WithTransaction(ctx, func(queries *dbs.Queries) error {
        fileID, err := queries.RttEmailFilesNew(ctx, dbs.RttEmailFilesNewParams{
            // file metadata
            CreatedAt: createdAt,
            CreatedBy: createdBy,
        })
        if err != nil {
            return fmt.Errorf("failed to create Rtt email file record: %v", err)
        }

        activatedIDs, err := queries.RttEmailsActivate(ctx, dbs.RttEmailsActivateParams{
            Emails:    emails,
            CreatedAt: createdAt,
        })
        if err != nil {
            return fmt.Errorf("failed to activate Rtt emails: %v", err)
        }

        if len(activatedIDs) > 0 {
            err := queries.RttEmailHistoryNew(ctx, dbs.RttEmailHistoryNewParams{
                EmailIds:  activatedIDs,
                Active:    true,
                FileID:    fileID,
                CreatedAt: createdAt,
            })
            if err != nil {
                return fmt.Errorf("failed to create Rtt email history records for activated emails: %v", err)
            }
        }

        deactivatedIDs, err := queries.RttEmailsDeactivate(ctx, dbs.RttEmailsDeactivateParams{
            UpdatedAt: createdAt,
            Emails:    emails,
        })
        if err != nil {
            return fmt.Errorf("failed to deactivate Rtt emails: %v", err)
        }

        if len(deactivatedIDs) > 0 {
            err := queries.RttEmailHistoryNew(ctx, dbs.RttEmailHistoryNewParams{
                EmailIds:  deactivatedIDs,
                Active:    false,
                FileID:    fileID,
                CreatedAt: createdAt,
            })
            if err != nil {
                return fmt.Errorf("failed to create Rtt email history records for deactivated emails: %v", err)
            }
        }

        activatedCount = int64(len(activatedIDs))
        deactivatedCount = int64(len(deactivatedIDs))
        return nil
    })
    if txErr != nil {
        return 0, 0, txErr
    }
    return activatedCount, deactivatedCount, nil
}

Processing 10k email addresses takes ~5 seconds; 100k already takes ~90 seconds. The problem is unnecessary round-trips to the database: the bulk approach makes separate queries for the UPSERT, the deactivation, and writing the history.

Optimized Approach (Single CTE Query)

The solution is to move all the logic into a single query using a CTE. Now the query looks like this:

-- name: RttEmailsSync :one
WITH
data (email) AS (
    SELECT UNNEST(@emails::VARCHAR[])
),
deactivated (id) AS (
    UPDATE rtt_emails
    SET active = FALSE, updated_at = @now::TIMESTAMP
    WHERE email NOT IN (SELECT email FROM data)
      AND active = TRUE
    RETURNING id
),
activated (id) AS (
    INSERT INTO rtt_emails (email, active, created_at, updated_at)
    SELECT email, TRUE, @now::TIMESTAMP, @now::TIMESTAMP
    FROM data
    ON CONFLICT (email) DO UPDATE
    SET active = EXCLUDED.active, updated_at = EXCLUDED.updated_at
    WHERE rtt_emails.active IS DISTINCT FROM EXCLUDED.active
    RETURNING id
),
deactivated_history AS (
    INSERT INTO rtt_email_history (email_id, active, file_id, created_at)
    SELECT id, FALSE, @file_id::BIGINT, @now::TIMESTAMP
    FROM deactivated
),
activated_history AS (
    INSERT INTO rtt_email_history (email_id, active, file_id, created_at)
    SELECT id, TRUE, @file_id::BIGINT, @now::TIMESTAMP
    FROM activated
)
SELECT (SELECT COUNT(*) FROM activated) AS activated_count,
       (SELECT COUNT(*) FROM deactivated) AS deactivated_count;

And here's what the insertion code looks like:

func (r *RttEmailRepository) UploadFast(
    ctx context.Context,
    file *FileMetadata,
    emails []string,
    createdAt time.Time,
    createdBy int64,
) (activatedCount int64, deactivatedCount int64, err error) {
    txErr := r.db.WithTransaction(ctx, func(queries *dbs.Queries) error {
        fileID, err := queries.RttEmailFilesNew(ctx, dbs.RttEmailFilesNewParams{
            // file metadata
            CreatedAt: createdAt,
            CreatedBy: createdBy,
        })
        if err != nil {
            return fmt.Errorf("failed to create Rtt email file record: %v", err)
        }

        result, err := queries.RttEmailsSync(ctx, dbs.RttEmailsSyncParams{
            Emails: emails,
            Now:    createdAt,
            FileID: fileID,
        })
        if err != nil {
            return fmt.Errorf("failed to sync Rtt emails: %v", err)
        }

        activatedCount = result.ActivatedCount
        deactivatedCount = result.DeactivatedCount
        return nil
    })
    if txErr != nil {
        return 0, 0, txErr
    }
    return activatedCount, deactivatedCount, nil
}

Processing 10k email addresses now takes ~3 seconds; 100k takes ~5 seconds; 1M takes ~30 seconds. For me, this kind of significant speedup - achieved by reducing the data transfer between the database and the application - was a real discovery, which is why I decided to write it up.

Epilogue

All of these examples came from real code I encountered during development. This article was originally written in Ukrainian this year - published on DOU. If you're working with Go in production and want to see which product companies use it, check out ReadyToTouch - Go companies.