Batch Upsert: COPY vs INSERT ON CONFLICT for pgvector

This page builds a head-to-head harness that measures ingestion throughput of three ways to land embedding batches in a pgvector table: binary COPY, multi-row INSERT ... ON CONFLICT DO UPDATE, and a COPY-into-staging-then-merge hybrid. It scopes the choice narrowly — append-only bulk load versus upsert-heavy refresh — and hands you a decision guide plus the timing and diagnostic queries to defend the pick.

Up: Batch Upsert Patterns for pgvector

The three strategies are not interchangeable. COPY ... FORMAT BINARY is the fastest way to push rows into Postgres, but it cannot resolve conflicts — a duplicate (doc_id, chunk_index) aborts the whole stream. Multi-row INSERT ... ON CONFLICT handles conflicts row-by-row but pays parse, plan, and per-row index-maintenance costs that dominate once the vector column carries an HNSW index. The hybrid — stream a batch into an UNLOGGED staging table with COPY, then run one set-based INSERT ... SELECT ... ON CONFLICT merge — recovers most of COPY’s bandwidth while still giving you upsert semantics. This page measures all three on the same rows so the trade-off is numbers, not folklore.

Prerequisites #

• PostgreSQL 15+ with the pgvector extension 0.5+ (CREATE EXTENSION vector;). PG15 is the floor because the MERGE alternative and improved ON CONFLICT planning matter to the comparison.
• psycopg 3.1+ (its cursor.copy() supports binary COPY) and optionally asyncpg 0.28+ whose copy_records_to_table is the async equivalent.
• numpy >= 1.24 and the pgvector Python package for the binary-COPY row encoder.
• A target table already typed vector(d) and, ideally, embeddings already L2-normalized per normalizing embeddings before pgvector insertion so the harness measures write cost, not normalization.
• A warmed connection: pool the harness through the same path production uses, sized per sizing asyncpg pools for embedding ingestion, so pool acquisition does not pollute the timings.

Step-by-step #

1. Create the target table and a matching staging table #

The target carries the HNSW index; the staging table is UNLOGGED, unindexed, and holds the same columns. UNLOGGED skips WAL for the staging writes, which is safe because the data is transient and re-derivable from the merge source.

CREATE TABLE IF NOT EXISTS doc_chunks (
    doc_id       text         NOT NULL,
    chunk_index  int          NOT NULL,
    embedding    vector(1536) NOT NULL,
    updated_at   timestamptz  DEFAULT now(),
    PRIMARY KEY (doc_id, chunk_index)
);
CREATE INDEX IF NOT EXISTS doc_chunks_hnsw
    ON doc_chunks USING hnsw (embedding vector_cosine_ops);

CREATE UNLOGGED TABLE IF NOT EXISTS doc_chunks_stage (
    doc_id       text         NOT NULL,
    chunk_index  int          NOT NULL,
    embedding    vector(1536) NOT NULL
);

2. Generate a repeatable batch #

Fix the random seed so every strategy sees identical rows and the comparison is apples-to-apples. Half the batch reuses existing keys to exercise the conflict path realistically.

import numpy as np
from pgvector.psycopg import register_vector

DIM, N = 1536, 50_000

def make_batch(seed: int = 7) -> list[tuple]:
    rng = np.random.default_rng(seed)
    vecs = rng.standard_normal((N, DIM)).astype(np.float32)
    vecs /= np.linalg.norm(vecs, axis=1, keepdims=True)
    rows = []
    for i in range(N):
        # first half are fresh doc_ids, second half collide with the first
        doc = f"doc-{i if i < N // 2 else i - N // 2}"
        rows.append((doc, i % 8, vecs[i]))
    return rows

3. Strategy A — binary COPY (append-only) #

COPY ... FORMAT BINARY streams rows with no per-row planning. Register the pgvector adapter so the vector is written in the compact binary wire format instead of a text literal.

import time, psycopg

def run_copy(conn, rows) -> float:
    register_vector(conn)
    t0 = time.perf_counter()
    with conn.cursor() as cur:
        with cur.copy(
            "COPY doc_chunks (doc_id, chunk_index, embedding) "
            "FROM STDIN WITH (FORMAT BINARY)"
        ) as cp:
            cp.set_types(["text", "int4", "vector"])
            for doc, idx, vec in rows:
                cp.write_row((doc, idx, vec))
    conn.commit()
    return time.perf_counter() - t0

This aborts the transaction if any key already exists. Use it only on a table you know is empty or partitioned by a fresh load window — a true append.

4. Strategy B — multi-row INSERT ON CONFLICT #

Batch many value tuples into one statement to amortize round-trips, but expect the index to be maintained once per row. Chunk at a few thousand rows per statement so the parameter count stays under Postgres’s 65,535 bind-parameter limit.

from psycopg import sql

def run_insert_on_conflict(conn, rows, page: int = 2000) -> float:
    register_vector(conn)
    t0 = time.perf_counter()
    with conn.cursor() as cur:
        for start in range(0, len(rows), page):
            chunk = rows[start:start + page]
            values = sql.SQL(",").join(sql.Placeholder() * 3 for _ in chunk)
            cur.execute(
                "INSERT INTO doc_chunks (doc_id, chunk_index, embedding) "
                "VALUES " + ",".join(["(%s,%s,%s)"] * len(chunk)) +
                " ON CONFLICT (doc_id, chunk_index) "
                "DO UPDATE SET embedding = EXCLUDED.embedding, updated_at = now()",
                [v for row in chunk for v in row],
            )
    conn.commit()
    return time.perf_counter() - t0

5. Strategy C — COPY into UNLOGGED staging, then merge #

Truncate staging, COPY the batch in at full bandwidth, then run one set-based INSERT ... SELECT ... ON CONFLICT. Deduplicate inside the staging table with DISTINCT ON so a batch that itself contains two rows for the same key does not trip ON CONFLICT’s “cannot affect row a second time” error.

def run_copy_merge(conn, rows) -> float:
    register_vector(conn)
    t0 = time.perf_counter()
    with conn.cursor() as cur:
        cur.execute("TRUNCATE doc_chunks_stage")
        with cur.copy(
            "COPY doc_chunks_stage (doc_id, chunk_index, embedding) "
            "FROM STDIN WITH (FORMAT BINARY)"
        ) as cp:
            cp.set_types(["text", "int4", "vector"])
            for doc, idx, vec in rows:
                cp.write_row((doc, idx, vec))
        cur.execute("""
            INSERT INTO doc_chunks (doc_id, chunk_index, embedding)
            SELECT DISTINCT ON (doc_id, chunk_index)
                   doc_id, chunk_index, embedding
            FROM doc_chunks_stage
            ORDER BY doc_id, chunk_index
            ON CONFLICT (doc_id, chunk_index)
            DO UPDATE SET embedding = EXCLUDED.embedding, updated_at = now()
        """)
    conn.commit()
    return time.perf_counter() - t0

6. Run the harness and compare #

Reset the table between runs so each strategy starts from the same state, and report rows per second.

def bench(conn):
    rows = make_batch()
    for name, fn in [("copy", run_copy),
                     ("insert_on_conflict", run_insert_on_conflict),
                     ("copy_merge", run_copy_merge)]:
        with conn.cursor() as cur:
            cur.execute("TRUNCATE doc_chunks")
        conn.commit()
        if name == "copy":               # append-only: needs an empty table
            fresh = [r for r in rows if r[0].endswith(tuple("01234"))]
            elapsed = fn(conn, fresh)
            n = len(fresh)
        else:
            elapsed = fn(conn, rows)
            n = len(rows)
        print(f"{name:20s} {n/elapsed:10.0f} rows/s  ({elapsed:.2f}s)")

As a heuristic, expect binary COPY to lead, the hybrid merge to land within roughly 10–30% of it, and row-by-row INSERT ... ON CONFLICT to trail by a wide margin once the HNSW index is present — the gap widens with dimension and index m. Treat these as directional; measure on your own hardware. For a fair read, keep the async ingestion path identical to production, described in async processing with Python asyncio.

Parameter reference #

Verification #

Confirm every strategy landed the same logical rows and no duplicates slipped in.

-- row count and duplicate check on the conflict key
SELECT count(*)                                        AS total_rows,
       count(*) - count(DISTINCT (doc_id, chunk_index)) AS dup_keys
FROM doc_chunks;

dup_keys must be 0. To confirm the merge actually updated rather than appended, compare counts before and after a re-run of the same batch: the total must not change on the second pass. Wall-clock timings from the harness plus this integrity check are the two numbers that justify the choice.

-- WAL generated by each strategy, sampled around a run
SELECT pg_current_wal_lsn();  -- diff two samples to size WAL per strategy

Troubleshooting #

• WAL and checkpoint storms during large INSERT ... ON CONFLICT. Symptom: throughput collapses periodically and pg_stat_bgwriter.checkpoints_timed climbs. Diagnostic: sample pg_current_wal_lsn() deltas per batch. Fix: route bulk loads through the UNLOGGED staging + merge path (strategy C), raise max_wal_size, and stagger commits.
• Per-row index maintenance dominates row-by-row INSERT. Symptom: INSERT ... ON CONFLICT is many times slower than COPY on the same rows. Diagnostic: EXPLAIN (ANALYZE, BUFFERS) shows most time under the HNSW insert. Fix: for append-heavy loads, drop the index, bulk COPY, then rebuild — coordinate the rebuild per scheduling REINDEX CONCURRENTLY without downtime.
• Deadlocks between concurrent upsert workers. Symptom: deadlock detected on doc_chunks. Diagnostic: pg_stat_activity shows two sessions each holding a row lock the other wants. Fix: sort each batch by the conflict key before insert so all workers acquire locks in the same order, and keep transactions short.
• ON CONFLICT cannot affect row a second time. Symptom: the merge aborts. Diagnostic: the batch (or staging table) contains two rows with the same (doc_id, chunk_index). Fix: apply the DISTINCT ON (conflict key) collapse shown in step 5, keeping the last write.
• Binary COPY fails with incorrect binary data format. Symptom: the stream aborts mid-batch. Diagnostic: the vector was written as a text literal, not binary. Fix: register_vector(conn) and cp.set_types([..., "vector"]) so psycopg emits the binary wire encoding; size the resulting storage with pgvector storage overhead analysis.

Related #

• Idempotent Vector Upserts with ON CONFLICT — the conflict-key mechanics behind the upsert strategies here
• Sizing asyncpg pools for embedding ingestion — keep pool acquisition out of your timings
• Normalizing embeddings before pgvector insertion — do this before the batch so you benchmark writes, not math
• Asynchronous index build strategies — when to drop-and-rebuild the HNSW index around a bulk load
• Up: Batch Upsert Patterns for pgvector