pgx COPY Protocol for Fast Bulk Loads Into Postgres

When the job is to get a lot of rows into one table, PostgreSQL has a protocol built for it, COPY, and pgx exposes it as CopyFrom. The data travels in a stream with almost no per statement overhead, and in binary form, so it is the fastest path from a Go program. This article uses it on PostgreSQL 18.6 from Go 1.25.6 with pgx v5.11.0: three ways to supply rows, what happens when something is wrong, how to get conflict handling that COPY lacks, and how to copy rows out.

It belongs to the database cornerstone. The comparison with other loading methods, with timings, is in bulk inserts, the batch alternative in pgx batches, and the conflict handling statement in upserts.

Three ways to feed it

pool.CopyFrom(ctx, pgx.Identifier{"table"}, columns, source) takes a table name (as an Identifier, so it is quoted properly), the column names and a CopyFromSource. pgx provides three sources.

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
)

const dsn = "postgres://postgres:[email protected]:55432/lab?sslmode=disable"

func must(err error) {
	if err != nil {
		log.Fatal(err)
	}
}

type reading struct {
	Sensor int32
	At     time.Time
	Value  float64
	Note   *string
}

func main() {
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, dsn)
	must(err)
	defer pool.Close()
	reset := func() {
		for _, q := range []string{"drop table if exists readings669", "create table readings669 (id bigserial primary key, sensor int not null, at timestamptz not null, value float8 not null, note text, unique (sensor, at))"} {
			_, err := pool.Exec(ctx, q)
			must(err)
		}
	}
	count := func() (n int) {
		_ = pool.QueryRow(ctx, "select count(*) from readings669").Scan(&n)
		return
	}
	cols := []string{"sensor", "at", "value", "note"}
	base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
	mk := func(i int) reading {
		var note *string
		if i%10 == 0 {
			s := "calibrating"
			note = &s
		}
		return reading{int32(i % 20), base.Add(time.Duration(i) * time.Second), float64(i) / 3, note}
	}

	// 1. CopyFromRows: a slice of rows already in memory.
	reset()
	rows := make([][]any, 20000)
	for i := range rows {
		r := mk(i)
		rows[i] = []any{r.Sensor, r.At, r.Value, r.Note}
	}
	start := time.Now()
	n, err := pool.CopyFrom(ctx, pgx.Identifier{"readings669"}, cols, pgx.CopyFromRows(rows))
	must(err)
	fmt.Printf("1. CopyFromRows: %d rows in %v, table holds %d\n", n, time.Since(start).Round(time.Millisecond), count())

	// 2. CopyFromSlice: convert each element when it is needed, without building [][]any first.
	reset()
	data := make([]reading, 20000)
	for i := range data {
		data[i] = mk(i)
	}
	start = time.Now()
	n, err = pool.CopyFrom(ctx, pgx.Identifier{"readings669"}, cols, pgx.CopyFromSlice(len(data), func(i int) ([]any, error) {
		return []any{data[i].Sensor, data[i].At, data[i].Value, data[i].Note}, nil
	}))
	must(err)
	fmt.Printf("2. CopyFromSlice: %d rows in %v\n", n, time.Since(start).Round(time.Millisecond))

	// 3. CopyFromFunc: rows produced one at a time, as when reading a file or a channel.
	reset()
	i := 0
	start = time.Now()
	n, err = pool.CopyFrom(ctx, pgx.Identifier{"readings669"}, cols, pgx.CopyFromFunc(func() ([]any, error) {
		if i >= 20000 {
			return nil, nil // a nil row ends the stream
		}
		r := mk(i)
		i++
		return []any{r.Sensor, r.At, r.Value, r.Note}, nil
	}))
	must(err)
	fmt.Printf("3. CopyFromFunc: %d rows in %v, streamed with no slice\n", n, time.Since(start).Round(time.Millisecond))

	// 4. One bad row fails the whole copy and loads nothing.
	reset()
	bad := append([][]any{}, rows[:1000]...)
	bad = append(bad, rows[0]) // the same (sensor, at) twice
	n, err = pool.CopyFrom(ctx, pgx.Identifier{"readings669"}, cols, pgx.CopyFromRows(bad))
	var pg *pgconn.PgError
	fmt.Printf("4. a duplicate in row 1001: loaded %d, rows in the table %d, error ", n, count())
	if errors.As(err, &pg) {
		fmt.Printf("%s (%s)\n", pg.Message, pg.Code)
	}

	// 5. A type that does not fit the column is caught on the client, before it is sent.
	reset()
	_, err = pool.CopyFrom(ctx, pgx.Identifier{"readings669"}, cols, pgx.CopyFromRows([][]any{{"not a number", base, 1.0, nil}}))
	fmt.Println("5. a string for an integer column:", err)

	// 6. COPY has no ON CONFLICT: load into a staging table, then merge.
	reset()
	// Temp tables belong to one session, so the staging steps share a connection.
	conn, err := pool.Acquire(ctx)
	must(err)
	defer conn.Release()
	_, err = conn.Exec(ctx, "create temp table stage669b (sensor int, at timestamptz, value float8, note text)")
	must(err)
	_, err = conn.CopyFrom(ctx, pgx.Identifier{"stage669b"}, cols, pgx.CopyFromRows(bad))
	must(err)
	tag, err := conn.Exec(ctx, "insert into readings669 (sensor, at, value, note) select distinct on (sensor, at) sensor, at, value, note from stage669b order by sensor, at on conflict (sensor, at) do nothing")
	must(err)
	fmt.Printf("6. staging table then insert ... on conflict do nothing: %s, table holds %d (the duplicate was dropped)\n", tag, count())

	// 7. Reading back out with COPY TO.
	var out []byte
	buf := &sliceWriter{&out}
	ct, err := conn.Conn().PgConn().CopyTo(ctx, buf, "copy (select sensor, value from readings669 where sensor = 3 order by at limit 2) to stdout with (format csv)")
	must(err)
	fmt.Printf("7. CopyTo wrote %d rows as CSV: %q\n", ct.RowsAffected(), string(out))
}

type sliceWriter struct{ b *[]byte }

func (w *sliceWriter) Write(p []byte) (int, error) { *w.b = append(*w.b, p...); return len(p), nil }

1. CopyFromRows: 20000 rows in 64ms, table holds 20000
2. CopyFromSlice: 20000 rows in 58ms
3. CopyFromFunc: 20000 rows in 45ms, streamed with no slice
4. a duplicate in row 1001: loaded 0, rows in the table 0, error duplicate key value violates unique constraint "readings669_sensor_at_key" (23505)
5. a string for an integer column: ERROR: COPY from stdin failed: unable to encode "not a number" into binary format for int4 (OID 23): cannot find encode plan (SQLSTATE 57014)
6. staging table then insert ... on conflict do nothing: INSERT 0 1000, table holds 1000 (the duplicate was dropped)
7. CopyTo wrote 2 rows as CSV: "3,1\n3,7.666666666666667\n"

CopyFromRows([][]any) takes rows already in memory as a slice of value slices. The lab built 20,000 rows and loaded them in 64ms. It is the simplest, and it holds a second copy of the data as []any.

CopyFromSlice(n, func(i int) ([]any, error)) takes the number of rows and a function that produces row i. Your data stays in its natural slice of structs, and each element is converted to []any only when it is needed: 58ms for the same rows. Use it when you have a slice of structs.

CopyFromFunc(func() ([]any, error)) calls the function repeatedly for the next row and stops when it returns nil. The lab produced its rows one at a time and took 45ms, never holding more than a row. Use it to stream from a file, a channel or another query, where building a slice would use too much memory. The differences in time are small and not the point; the point is that the function form needs no slice at all.

Values are written in PostgreSQL's binary format using the same type mapping as other pgx calls. Times, floats and nullable values (a *string that is nil becomes NULL) work as you expect, and the table's columns must be listed in the order the values are given.

All or nothing

COPY loads inside one implicit transaction, and the fourth block shows the consequence. The program sent 1,000 good rows followed by a copy of the first row, so the 1,001st violated the unique constraint on (sensor, at). The result: loaded 0, rows in the table 0, with the error duplicate key value violates unique constraint "readings669_sensor_at_key" and code 23505. A thousand valid rows were not loaded because of one bad one. The error does not say which row was at fault, only which constraint, so for untrusted data validate in Go first, or load into a staging table and clean it there.

The fifth block sends a string where the column wants an integer. The error came back as COPY from stdin failed: unable to encode "not a number" into binary format for int4 (OID 23): cannot find encode plan, with SQLSTATE 57014. pgx detected the problem while encoding on the client, aborted the copy, and PostgreSQL reported the abort. As before, nothing was loaded.

COPY has no ON CONFLICT

The statement COPY cannot skip duplicates or update existing rows. The standard workaround is a staging table: copy into a temporary table that has no constraints, then merge with SQL.

The sixth block does that. It takes one connection from the pool, because a temporary table belongs to a session, creates stage669b without constraints, copies 1,001 rows into it (including the duplicate) and runs:

insert into readings (sensor, at, value, note) select distinct on (sensor, at) sensor, at, value, note from stage order by sensor, at on conflict (sensor, at) do nothing

The tag was INSERT 0 1000 and the table held 1,000 rows: the duplicate was dropped by distinct on, and anything that already existed in the target would have been skipped by on conflict. Replace do nothing with do update set ... for an upsert. The staging table costs a second write of the data, which is the price of having SQL available between loading and the final table, and it is still far faster than row by row inserts. It is also the place to cast, validate, join against reference tables and reject bad rows with their keys. Drop the staging table after use (a temp table goes away with the session; with a pool the session lives on, so drop table or on commit drop inside a transaction).

Copying out

COPY also works in the other direction. The last block uses the lower level connection, conn.Conn().PgConn().CopyTo(ctx, writer, sql), to stream the result of copy (select ...) to stdout with (format csv) into an io.Writer: the command tag said 2 rows, and the bytes were 3,1 and 3,7.666666666666667 as CSV. Use it for exports: it streams, so a table of any size flows to a file or an HTTP response without memory growing. Pass a with (format csv, header) option for a header line. pgx's CopyFrom reads from Go values; CopyTo produces text or CSV bytes, which you parse yourself if you need the values.

Practical rules

1. Use CopyFrom for thousands of rows or more into one table. For a few hundred rows, a batch is simpler and fast enough.
2. Stream with CopyFromFunc when the data is large or arrives incrementally.
3. Expect all or nothing. Plan for a bad row: validate beforehand or use a staging table.
4. Use a staging table for conflicts, and do the merge in one SQL statement.
5. Chunk very large loads into several copies of, say, 100,000 rows if a single transaction would be too large to retry; keep track of which chunks finished.
6. Know the limits: COPY does not return generated values (no returning), it fires triggers and checks constraints (so a heavily indexed table loads slower; drop indexes for an initial load and create them after), and it needs the types to match the column types exactly.
7. Set a deadline on the context for the whole load, and treat cancellation as an abort.

The protocol is documented in PostgreSQL's page on COPY, and the pgx functions in the package reference.