Module 7: Bulk Operations

Bulk upserts with a temp table: the canonical pattern

You have two requirements: speed (10k+ rows) and idempotency (tolerate duplicates). COPY gives you speed. ON CONFLICT gives you idempotency. But direct COPY doesn't support ON CONFLICT. How do you combine them?

The answer is the canonical bulk upsert pattern: COPY into a temp table, then INSERT ... SELECT ... ON CONFLICT from the temp into the real table. Three steps, fully atomic, scales to millions of rows.

This is the pattern any production PostgreSQL ETL uses. Once you know it, your intuition for "large import with possible duplicates" goes from "I'll iterate one by one with try/except" to "COPY into temp + INSERT SELECT ON CONFLICT". The difference between hours and seconds.

The pattern in pure SQL

-- 1. Create a temp table with the same structure as the destination
CREATE TEMP TABLE tmp_tasks_import (
    external_id TEXT,
    title TEXT,
    status TEXT,
    priority INT,
    synced_at TIMESTAMPTZ
) ON COMMIT DROP;

-- 2. COPY records into the temp (fast, no strict constraints)
COPY tmp_tasks_import FROM STDIN WITH (FORMAT csv);
-- ... data via STDIN ...
\.

-- 3. INSERT from temp into the real table with ON CONFLICT
INSERT INTO tasks (external_id, title, status, priority, synced_at)
SELECT external_id, title, status, priority, synced_at
FROM tmp_tasks_import
ON CONFLICT (external_id) DO UPDATE SET
    title = EXCLUDED.title,
    status = EXCLUDED.status,
    priority = EXCLUDED.priority,
    synced_at = EXCLUDED.synced_at;

-- 4. The temp table is dropped automatically on COMMIT (because of ON COMMIT DROP)

ON COMMIT DROP ensures the temp table is cleaned up at the end of the transaction. Without it, temp tables would accumulate.

Implementation in Python

async def bulk_upsert_with_temp(
    records: list[tuple],
    conn: asyncpg.Connection,
):
    async with conn.transaction():
        # 1. Create temp table
        await conn.execute("""
            CREATE TEMP TABLE tmp_tasks_import (
                external_id TEXT,
                title TEXT,
                status TEXT,
                priority INT,
                synced_at TIMESTAMPTZ
            ) ON COMMIT DROP
        """)

        # 2. COPY into temp (ultra-fast)
        await conn.copy_records_to_table(
            "tmp_tasks_import",
            records=records,
            columns=["external_id", "title", "status", "priority", "synced_at"],
        )

        # 3. INSERT...SELECT...ON CONFLICT from temp into the real table
        result = await conn.fetch("""
            INSERT INTO tasks (external_id, title, status, priority, synced_at)
            SELECT external_id, title, status, priority, synced_at
            FROM tmp_tasks_import
            ON CONFLICT (external_id) DO UPDATE SET
                title = EXCLUDED.title,
                status = EXCLUDED.status,
                priority = EXCLUDED.priority,
                synced_at = EXCLUDED.synced_at
            RETURNING id, (xmax = 0) AS inserted
        """)

        inserted = sum(1 for r in result if r["inserted"])
        updated = len(result) - inserted

        return {"inserted": inserted, "updated": updated, "total": len(result)}

async with conn.transaction() wraps everything. If any step fails, automatic rollback.

Why it's so efficient

The technical reasons:

1. COPY into temp is maximum speed. The temp table has no constraints (UNIQUE, FK), no triggers (unless you add them), no indexes. Raw COPY into a simple table is as fast as it gets.

2. INSERT SELECT reads from temp in optimal order. PostgreSQL can plan the INSERT...SELECT as a Hash Join between the temp and the destination table, using the UNIQUE constraint's index for the ON CONFLICT check. Much more efficient than processing row-by-row from the app.

3. A single transaction. COPY + INSERT within the same transaction. There are no intermediate round-trips to the app. PostgreSQL does everything internally.

4. ON COMMIT DROP frees space immediately. The temp table disappears on commit. No cleanup overhead left behind.

Comparative performance

For 100k rows with 30% duplicates:

Approach	Time	Throughput
Loop INSERT with try/except UPDATE	~120s	800/s
`pg_insert(...).on_conflict_do_update(...)` with bulk values	~6s	16k/s
COPY → temp → INSERT...ON CONFLICT	~1.5s	66k/s

For 1M rows:

Approach	Time
Loop with try/except	OOM or timeout
Bulk pg_insert	~60s
Temp pattern	~14s

The temp pattern scales much better to large volumes.

Variants of the pattern

Validation in SQL before the INSERT

async def upsert_with_validation(records, conn):
    async with conn.transaction():
        await conn.execute("""
            CREATE TEMP TABLE tmp_import (
                external_id TEXT,
                title TEXT,
                status TEXT,
                priority INT,
                valid BOOL DEFAULT TRUE,
                error_msg TEXT
            ) ON COMMIT DROP
        """)

        await conn.copy_records_to_table(
            "tmp_import",
            records=records,
            columns=["external_id", "title", "status", "priority"],
        )

        # Validate in SQL — faster than iterating in Python
        await conn.execute("""
            UPDATE tmp_import SET valid = FALSE, error_msg = 'invalid status'
            WHERE status NOT IN ('pending', 'in_progress', 'completed', 'archived')
        """)

        await conn.execute("""
            UPDATE tmp_import SET valid = FALSE, error_msg = 'priority out of range'
            WHERE priority < 1 OR priority > 5
        """)

        await conn.execute("""
            UPDATE tmp_import SET valid = FALSE, error_msg = 'title too long'
            WHERE LENGTH(title) > 200
        """)

        # INSERT only valid rows
        inserted = await conn.fetch("""
            INSERT INTO tasks (external_id, title, status, priority)
            SELECT external_id, title, status, priority
            FROM tmp_import
            WHERE valid = TRUE
            ON CONFLICT (external_id) DO UPDATE SET
                title = EXCLUDED.title,
                status = EXCLUDED.status,
                priority = EXCLUDED.priority
            RETURNING id
        """)

        # Report errors
        errors = await conn.fetch("""
            SELECT external_id, error_msg
            FROM tmp_import
            WHERE NOT valid
        """)

        return {
            "imported": len(inserted),
            "errors": [dict(e) for e in errors],
            "skipped": len(errors),
        }

This variant allows partial success: skipping invalid rows instead of failing everything. Useful for imports where some errors are expected.

Filter by timestamp (latest only)

If your source has multiple versions of the same record and you want only the most recent one:

INSERT INTO tasks (external_id, title, status, version)
SELECT DISTINCT ON (external_id) external_id, title, status, version
FROM tmp_import
ORDER BY external_id, version DESC
ON CONFLICT (external_id) DO UPDATE SET ...
WHERE tasks.version < EXCLUDED.version;

DISTINCT ON (external_id) ... ORDER BY ... version DESC keeps only the row with the max version per external_id. Combined with WHERE version < EXCLUDED.version in the ON CONFLICT, it's perfect for syncing events with partial ordering.

Lookup in another table

If you need to validate that external_user_id exists in users:

INSERT INTO tasks (external_id, title, owner_id)
SELECT
    t.external_id,
    t.title,
    u.id AS owner_id
FROM tmp_import t
INNER JOIN users u ON u.external_id = t.external_user_id
ON CONFLICT (external_id) DO UPDATE SET ...;
-- Rows with no matching user in `users` simply aren't inserted (filtered out)

Streaming for gigantic datasets

For datasets >1M rows, don't load everything into Python memory. Stream from a generator:

async def stream_upsert(record_generator, conn):
    async with conn.transaction():
        await conn.execute("""
            CREATE TEMP TABLE tmp_import (...) ON COMMIT DROP
        """)

        # COPY from generator (streaming)
        await conn.copy_records_to_table(
            "tmp_import",
            records=record_generator,  # generator, not a list
            columns=[...],
        )

        # INSERT SELECT from temp
        await conn.execute("""
            INSERT INTO tasks ... SELECT ... FROM tmp_import
            ON CONFLICT (...) DO UPDATE ...
        """)


def parse_huge_csv():
    """Generator that yields records from a large file."""
    with open("huge.csv") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield (
                row["external_id"],
                row["title"],
                row["status"],
                int(row["priority"]),
            )


async def import_huge():
    conn = await asyncpg.connect(...)
    result = await stream_upsert(parse_huge_csv(), conn)
    await conn.close()

Memory kept under control: only one row in Python memory at a time (asyncpg internally uses a small buffer).

Complete endpoint with stats

import time
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel


router = APIRouter()


class TaskBulkData(BaseModel):
    external_id: str
    title: str
    status: str
    priority: int


class BulkUpsertResponse(BaseModel):
    inserted: int
    updated: int
    skipped: int
    errors: list[dict]
    elapsed_ms: int


@router.post("/tasks/bulk-upsert", response_model=BulkUpsertResponse)
async def bulk_upsert_endpoint(
    tasks: list[TaskBulkData],
    db: AsyncSession = Depends(get_db),
):
    if len(tasks) > 100_000:
        raise HTTPException(
            status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail="Max 100,000 tasks per request"
        )

    start = time.perf_counter()

    records = [
        (t.external_id, t.title, t.status, t.priority)
        for t in tasks
    ]

    raw_conn = await db.connection()
    asyncpg_conn = await raw_conn.get_raw_connection()
    pg_conn = asyncpg_conn.driver_connection

    async with pg_conn.transaction():
        # Setup
        await pg_conn.execute("""
            CREATE TEMP TABLE tmp_tasks_import (
                external_id TEXT,
                title TEXT,
                status TEXT,
                priority INT,
                valid BOOL DEFAULT TRUE,
                error_msg TEXT
            ) ON COMMIT DROP
        """)

        # COPY
        await pg_conn.copy_records_to_table(
            "tmp_tasks_import",
            records=records,
            columns=["external_id", "title", "status", "priority"],
        )

        # Validate
        await pg_conn.execute("""
            UPDATE tmp_tasks_import SET valid = FALSE, error_msg = 'invalid status'
            WHERE status NOT IN ('pending', 'in_progress', 'completed', 'archived')
        """)

        # Upsert valid rows
        result = await pg_conn.fetch("""
            INSERT INTO tasks (external_id, title, status, priority)
            SELECT external_id, title, status, priority FROM tmp_tasks_import WHERE valid = TRUE
            ON CONFLICT (external_id) DO UPDATE SET
                title = EXCLUDED.title,
                status = EXCLUDED.status,
                priority = EXCLUDED.priority
            RETURNING id, (xmax = 0) AS inserted
        """)

        # Errors
        errors = await pg_conn.fetch("""
            SELECT external_id, error_msg FROM tmp_tasks_import WHERE NOT valid
        """)

    inserted = sum(1 for r in result if r["inserted"])
    updated = len(result) - inserted

    elapsed_ms = int((time.perf_counter() - start) * 1000)

    return BulkUpsertResponse(
        inserted=inserted,
        updated=updated,
        skipped=len(errors),
        errors=[dict(e) for e in errors],
        elapsed_ms=elapsed_ms,
    )

Test it:

curl -X POST http://localhost:8000/tasks/bulk-upsert \
  -H "Content-Type: application/json" \
  -d '[
    {"external_id": "ext-1", "title": "T1", "status": "pending", "priority": 5},
    {"external_id": "ext-1", "title": "T1 dup", "status": "pending", "priority": 5},
    {"external_id": "ext-2", "title": "T2", "status": "invalid_status", "priority": 3}
  ]'

# Response:
# {
#   "inserted": 1,    # ext-1 (the first version wins)
#   "updated": 0,
#   "skipped": 1,     # ext-2 invalid status
#   "errors": [{"external_id": "ext-2", "error_msg": "invalid status"}],
#   "elapsed_ms": 12
# }

Pitfalls and common mistakes

1. Forgetting ON COMMIT DROP.

Without it, the temp table stays until the connection closes. If you use connection pooling, gradual accumulation.

2. CREATE TEMP TABLE outside a transaction.

If your connection pool reuses connections, the temp table can be seen by the next transaction. ON COMMIT DROP cleans it up, but only if it's inside a transaction.

3. Forgetting async with conn.transaction().

Without an explicit transaction, each statement commits automatically. The temp table can be partially visible, COPY can commit without the INSERT, etc.

4. INSERT SELECT without filtering invalid rows.

If you validate in temp but the INSERT SELECT doesn't use WHERE valid = TRUE, the invalid rows are attempted against the real table (causing constraint errors).

5. Too much validation in Python instead of SQL.

# ❌ Slow
for record in records:
    if validate_status(record):  # Python check
        valid_records.append(record)

# ✅ Fast
# COPY all to temp, then UPDATE temp SET valid = FALSE WHERE ...

Vectorized SQL is orders of magnitude faster than Python iteration.

6. Schema mismatch between temp and real.

If the temp table has different types than the real one (e.g., TEXT in temp vs VARCHAR(50) in real), INSERT SELECT can fail. Use the same schema or an explicit cast.

7. No explicit commit.

If the transaction is manual and you forget to commit, ON COMMIT DROP never fires and the temp table isn't dropped.

8. Timeouts on long transactions.

For datasets of millions, the whole transaction can take minutes. Check that statement_timeout and idle_in_transaction_session_timeout don't abort it.

Exercise: implement the complete pipeline

Setup: a Task model with external_id UNIQUE.

class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    external_id: Mapped[str] = mapped_column(String(100), unique=True)
    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[str] = mapped_column(String(50))
    priority: Mapped[int] = mapped_column(Integer)

Step 1: implement the endpoint with the temp pattern.

(Code in the "Complete endpoint" section above)

Step 2: test with a small dataset (5 records, 1 duplicate, 1 invalid).

data = [
    {"external_id": "e1", "title": "T1", "status": "pending", "priority": 5},
    {"external_id": "e2", "title": "T2", "status": "completed", "priority": 3},
    {"external_id": "e3", "title": "T3", "status": "pending", "priority": 1},
    {"external_id": "e1", "title": "T1 updated", "status": "in_progress", "priority": 5},  # duplicate, will overwrite
    {"external_id": "e4", "title": "T4", "status": "BAD", "priority": 5},  # invalid
]

# Expected result:
# inserted: 3 (e1, e2, e3 — but e1 with the value from the second appearance)
# updated: 0 (nothing existed before)
# skipped: 1 (e4)

Wait: what happens with e1, which appears twice? The COPY puts both into temp. The INSERT SELECT with ON CONFLICT only inserts one (the first one, by something random — the order of the temp table). To guarantee the behavior, use DISTINCT ON or filter beforehand.

Step 3: guarantee "latest wins" with DISTINCT ON.

# Change the INSERT to:
"""
INSERT INTO tasks (...)
SELECT DISTINCT ON (external_id) external_id, title, status, priority
FROM tmp_tasks_import
WHERE valid = TRUE
ORDER BY external_id, /* some column that defines the order, e.g.: synced_at DESC */
ON CONFLICT (external_id) DO UPDATE SET ...
"""

Step 4: benchmark with 100k records.

records = generate_data(100_000)  # with 30% duplicates
elapsed_ms = await bulk_upsert(records)
print(f"100k records (30% dupes): {elapsed_ms}ms")
# Expected: ~1500-2500ms

Step 5: stream for 10M records.

def generate_huge():
    for i in range(10_000_000):
        yield (f"ext-{i}", f"T{i}", "pending", i % 5)


await stream_upsert(generate_huge(), conn)
# Expected: ~3-5 minutes for 10M rows with upserts

See discussion

Step 2 — observation:

e1 appears twice in the COPY → both go into temp. INSERT SELECT with ON CONFLICT inserts the first occurrence, and for the second it triggers ON CONFLICT (because it's already in tasks). If DO UPDATE, the second one "wins" (overwrites). But the order depends on PostgreSQL — not guaranteed.

Step 3 — guarantee order:

DISTINCT ON (external_id) ... ORDER BY external_id, sync_timestamp DESC keeps only the most recent one. Combine it with WHERE in ON CONFLICT for deterministic "last write wins".

Step 4 — benchmark:

100k records with 30% dupes: ~1.5-2.5s. The bulk of it is the COPY into temp (faster) and the INSERT SELECT with the merge (slower, because of the constraint check).

Step 5 — streaming at scale:

10M records with a generator: ~3-5 minutes. Python memory <100MB throughout the whole process. PostgreSQL accumulates the data in temp + processes it in batch.

Key takeaways:

The temp-table pattern is constant for any scale (small or large).
Validation in SQL is orders of magnitude faster than Python iteration.
Streaming with a generator is necessary for >>1M records.
DISTINCT ON is the way to handle duplicates in the source.

Summary and next step

What you learned:

Canonical bulk upsert pattern: COPY into temp → INSERT...SELECT...ON CONFLICT from temp.
Steps: CREATE TEMP TABLE ... ON COMMIT DROP → copy_records_to_table → INSERT...SELECT...ON CONFLICT.
Performance: for 100k rows with duplicates, ~1.5s. For 10M streaming, ~3-5min.
Validation in SQL (UPDATE temp SET valid = FALSE WHERE ...) > validation in Python.
Streaming with generators for gigantic datasets.
DISTINCT ON for handling duplicates in the source.
Pitfalls: forgetting ON COMMIT DROP, not using an explicit transaction, validating in Python instead of SQL.

Before moving on, you should be able to:

Implement the complete pattern in pure SQL and in Python.
Decide when to add a validation step in SQL vs Python.
Handle duplicates in the source with DISTINCT ON.
Stream for datasets that don't fit in memory.

In the next capsule we cover the edge cases: robust error handling and streaming for gigantic datasets. How to tell recoverable errors from fatal ones, how to report progress to the client during long imports, and patterns to avoid OOM when you import millions of rows.

Resources

PostgreSQL Docs — Temp tables — reference for TEMP tables and ON COMMIT.
Citus Data — Faster bulk loading with COPY — specific patterns.
Heap — Idempotent ETL — patterns at scale with temp tables.
Brandur Leach — Postgres bulk operations — deep dive.
PostgreSQL Wiki — UPSERT performance — performance tips.
Mike Bayer — SQLAlchemy bulk patterns — official examples.

Capsule 06 of 08 — Module 7 — SQL Patterns for Production APIs Guide

← PreviousAtomic upserts with `ON CONFLICT`Next →Robust error handling + streaming for gigantic datasets