sql python 75 lines · 3 tabs

Batched writes with COPY (conceptual)

Shared by codesnips Jan 2026

3 tabs

CREATE TABLE events (
    id          bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    source      text        NOT NULL,
    external_id text        NOT NULL,
    payload     jsonb       NOT NULL,
    occurred_at timestamptz NOT NULL,
    updated_at  timestamptz NOT NULL DEFAULT now(),
    CONSTRAINT events_natural_key UNIQUE (source, external_id)
);

CREATE INDEX events_occurred_at_idx ON events (occurred_at);

BEGIN;

CREATE TEMP TABLE events_staging (
    source      text,
    external_id text,
    payload     jsonb,
    occurred_at timestamptz
) ON COMMIT DROP;

-- Streamed from the client; the fastest way to get rows into Postgres.
COPY events_staging (source, external_id, payload, occurred_at)
FROM STDIN WITH (FORMAT csv);

INSERT INTO events (source, external_id, payload, occurred_at)
SELECT DISTINCT ON (source, external_id)
       source, external_id, payload, occurred_at
FROM events_staging
ORDER BY source, external_id, occurred_at DESC
ON CONFLICT (source, external_id) DO UPDATE
    SET payload     = excluded.payload,
        occurred_at = excluded.occurred_at,
        updated_at  = now()
    WHERE events.payload IS DISTINCT FROM excluded.payload;

COMMIT;

import json
import psycopg

STAGE = """
CREATE TEMP TABLE events_staging (
    source text, external_id text, payload jsonb, occurred_at timestamptz
) ON COMMIT DROP
"""

MERGE = """
INSERT INTO events (source, external_id, payload, occurred_at)
SELECT DISTINCT ON (source, external_id)
       source, external_id, payload, occurred_at
FROM events_staging
ORDER BY source, external_id, occurred_at DESC
ON CONFLICT (source, external_id) DO UPDATE
    SET payload = excluded.payload,
        occurred_at = excluded.occurred_at,
        updated_at = now()
    WHERE events.payload IS DISTINCT FROM excluded.payload
"""


def load_events(dsn, rows):
    with psycopg.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute(STAGE)
            copy_sql = (
                "COPY events_staging (source, external_id, payload, occurred_at) "
                "FROM STDIN"
            )
            with cur.copy(copy_sql) as copy:
                for r in rows:
                    copy.write_row(
                        (r["source"], r["external_id"],
                         json.dumps(r["payload"]), r["occurred_at"])
                    )
            cur.execute(MERGE)
        conn.commit()

3 files · sql, python Explain with highlit

Loading millions of rows one INSERT at a time is dominated by per-statement overhead: parsing, planning, WAL flushes, and round trips. Postgres offers a much faster path with COPY, which streams rows in a compact protocol and writes them in large batches. These files show the standard pattern for a reliable bulk load that also handles duplicates and conflicting updates, since raw COPY alone cannot express ON CONFLICT.

The first tab, schema.sql, defines the durable target events table with a natural key on source plus external_id, backed by a UNIQUE constraint. That constraint is what makes the later upsert idempotent: re-running a load with overlapping rows will not create duplicates. It also adds an updated_at column so the merge step can record when a row last changed.

The second tab, load_events.sql, is the core of the batched-write technique. It creates an UNLOGGED staging table with CREATE TEMP TABLE ... ON COMMIT DROP, which skips WAL entirely for the scratch data. Rows are streamed in with COPY events_staging FROM STDIN, the fastest ingestion primitive in Postgres. Once the raw data lands, a single INSERT ... SELECT ... ON CONFLICT DO UPDATE merges the whole batch into events in one set-based statement. The DISTINCT ON (source, external_id) guards against duplicates within the same batch, which ON CONFLICT cannot resolve on its own because a command cannot hit the same conflict target twice. The WHERE events.payload IS DISTINCT FROM excluded.payload clause avoids no-op writes, so unchanged rows do not generate dead tuples or bump updated_at.

The third tab, bulk_load.py, drives the flow from an application using psycopg. It wraps the temp-table creation, the copy stream, and the merge in one transaction so a failure rolls back cleanly and leaves events untouched. Using cursor.copy with binary-safe row iteration keeps memory flat regardless of file size.

The trade-off is that data briefly lives in an unindexed staging area, and the merge scans it fully, so extremely small batches are not worth the ceremony. For large loads the win is dramatic. This pattern is the go-to for ETL, backfills, and ingesting external feeds where both speed and idempotent upserts matter.

Related snips

package dbutil

import (
  "context"

  "github.com/jackc/pgconn"

Retry Postgres serialization failures with bounded attempts

go postgres transactions

by Leah Thompson 1 tab

ruby

require "csv"

class PeopleCsvStream
  include Enumerable

  HEADERS = %w[id full_name email signed_up_at plan].freeze

Resilient CSV Export as a Streamed Response

rails performance streaming

by codesnips 3 tabs

ruby

Rails.application.configure do
  config.after_initialize do
    Bullet.enable = true
    Bullet.alert = false
    Bullet.bullet_logger = true
    Bullet.console = true

N+1 query detection with Bullet gem

rails performance activerecord

by Alex Kumar 2 tabs

ruby

# BAD: N+1 query problem
@users = User.all
@users.each do |user|
  puts user.posts.count  # Fires query for each user!
end

ActiveRecord query optimization and N+1 prevention

ruby rails activerecord

by Sarah Mitchell 3 tabs

ruby

json.array! @posts do |post|
  json.cache! ['v1', post], expires_in: 1.hour do
    json.id post.id
    json.title post.title
    json.excerpt post.excerpt
    json.published_at post.published_at

Fragment caching for expensive JSON serialization

rails caching performance

by Alex Kumar 1 tab

ruby

class AddSettingsToAccounts < ActiveRecord::Migration[7.1]
  disable_ddl_transaction!

  def change
    add_column :accounts, :settings, :jsonb, null: false, default: {}

Postgres JSONB Partial Index for Feature Flags

rails postgres jsonb

by codesnips 3 tabs

Share this code

Here's the card — post it anywhere.

Batched writes with COPY (conceptual) — share card

Link copied