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.
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
);
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. 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.
Top comments (0)