Mastering PostgreSQL CSV Imports: COPY, \copy, and Beyond

PGHOST=127.0.0.1
PGPORT=5432
PGDATABASE=testdb
PGUSER=postgres
PGPASSWORD=testpass

COPY customers (full_name, email, signup_source, created_at)
FROM '/datasets/large.csv'
WITH (
  FORMAT CSV,
  HEADER,
  DELIMITER ',',
  ENCODING 'UTF8'
);

PGPASSWORD=$PGPASSWORD \
psql -h $PGHOST -U $PGUSER -d $PGDATABASE \
  -c "\\copy customers (full_name, email, signup_source, created_at) \
        FROM '/path/to/datasets/small.csv' \
        WITH (FORMAT CSV, HEADER)"

const { Client } = require('pg');
const copyFrom = require('pg-copy-streams').from;
const { createReadStream } = require('node:fs');

const csvPath = process.env.CSV_PATH ?? 'datasets/medium.csv';

async function run() {
  const client = new Client({ connectionString: process.env.DATABASE_URL });
  await client.connect();
  try {
    const stream = client.query(
      copyFrom(
        `COPY customers (full_name, email, signup_source, created_at)
       FROM STDIN WITH (FORMAT CSV, HEADER)`,
      ),
    );

    await new Promise((resolve, reject) => {
      createReadStream(csvPath)
        .on('error', reject)
        .pipe(stream)
        .on('finish', resolve)
        .on('error', reject);
    });
  } finally {
    await client.end();
  }
}

run().catch((err) => {
  console.error('Import failed', err);
  process.exitCode = 1;
});

import os
import psycopg

conn = psycopg.connect(os.environ["DATABASE_URL"], autocommit=False)

with conn, conn.cursor() as cur:
    with open("datasets/medium.csv", "r", encoding="utf-8") as infile:
        cur.copy_expert(
            """
            COPY customers (full_name, email, signup_source, created_at)
            FROM STDIN WITH (FORMAT CSV, HEADER)
            """,
            infile,
        )
    cur.execute("ANALYZE customers;")

package main

import (
    "context"
    "encoding/csv"
    "io"
    "log"
    "os"

    "github.com/jackc/pgx/v5"
)
func main() {
    ctx := context.Background()
    conn, err := pgx.Connect(ctx, os.Getenv("DATABASE_URL"))
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close(ctx)

    rows := make(chan []any)
    go func() {
        defer close(rows)

        csvPath := os.Getenv("CSV_PATH")
        if csvPath == "" {
            csvPath = "datasets/medium.csv"
        }
        file, err := os.Open(csvPath)
        if err != nil {
            log.Fatal(err)
        }
        defer file.Close()

        reader := csv.NewReader(file)
        if _, err := reader.Read(); err != nil {
            log.Fatal(err) // skip header
        }
        for {
            record, err := reader.Read()
            if err == io.EOF {
                return
            }
            if err != nil {
                log.Fatal(err)
            }
            rows <- []any{record[0], record[1], record[2], record[3]}
        }
    }()

    _, err = conn.CopyFrom(
        ctx,
        pgx.Identifier{"customers"},
        []string{"full_name", "email", "signup_source", "created_at"},
        pgx.CopyFromChannel(rows),
    )
    if err != nil {
        log.Fatal(err)
    }
}

CREATE UNLOGGED TABLE staging.customers_import (
  full_name TEXT,
  email TEXT,
  signup_source TEXT,
  created_at TIMESTAMPTZ
);

COPY staging.customers_import FROM '/datasets/medium.csv'
WITH (FORMAT CSV, HEADER);

INSERT INTO public.customers (full_name, email, signup_source, created_at)
SELECT DISTINCT full_name, email, signup_source, created_at
FROM staging.customers_import
ON CONFLICT (email) DO UPDATE
  SET signup_source = EXCLUDED.signup_source,
      created_at = EXCLUDED.created_at;

TRUNCATE staging.customers_import;

COPY customers (full_name, email, signup_source, created_at)
FROM PROGRAM 'curl -s https://example.com/export.csv'
WITH (FORMAT CSV, HEADER);

SELECT * FROM pg_stat_progress_copy;

#!/usr/bin/env python3
"""Generate reproducible CSV datasets for PostgreSQL COPY testing."""
from __future__ import annotations

import argparse
import csv
import random
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Callable, Iterable, Sequence

DEFAULT_HEADERS = ["full_name", "email", "signup_source", "created_at"]
DEFAULT_SOURCES = ["referral", "website", "product-hunt", "conference", "webinar"]


@dataclass
class Profile:
    name: str
    headers: Sequence[str]
    delimiter: str = ","
    encoding: str = "utf-8"
    row_factory: Callable[[random.Random, int], Iterable[str]] | None = None
    uses_csv_writer: bool = True
    custom_writer: Callable[[Path], None] | None = None


def base_row_factory(rng: random.Random, index: int) -> Iterable[str]:
    first_names = ["Anne", "Carlos", "Dev", "Fatima", "Lina", "Mason", "Priya", "Yara"]
    last_names = ["Ng", "Rodriguez", "Iyer", "Hughes", "Ono", "Dubois", "Singh", "Owens"]
    full_name = f"{rng.choice(first_names)} {rng.choice(last_names)}"
    email = f"{full_name.lower().replace(' ', '.')}+{index}@example.com"
    signup_source = rng.choice(DEFAULT_SOURCES)
    created_at = datetime(2024, 1, 1) + timedelta(minutes=rng.randint(0, 525600))
    return [full_name, email, signup_source, created_at.isoformat(sep=" ")]


def quoted_commas_factory(rng: random.Random, index: int) -> Iterable[str]:
    base = list(base_row_factory(rng, index))
    base[0] = f'{base[0]} "The Third"'
    base[2] = f"conference, booth {rng.randint(1, 99)}"
    return base


def multiline_factory(rng: random.Random, index: int) -> Iterable[str]:
    name, email, _, created_at = base_row_factory(rng, index)
    address_lines = [
        f"{rng.randint(100, 999)} Market Street",
        f"Suite {rng.randint(1, 50)}",
        f"Seattle, WA {rng.randint(90000, 99999)}"
    ]
    return [name, email, "web-form", created_at, "\n".join(address_lines)]


def latin1_factory(rng: random.Random, index: int) -> Iterable[str]:
    names = ["Álvaro Nieves", "Chloé Martin", "Mårten Björk", "Señora Núñez", "Zoë Brontë"]
    full_name = rng.choice(names)
    email = f"latin{index}@example.com"
    signup_source = rng.choice(DEFAULT_SOURCES)
    created_at = datetime(2024, 6, 1) + timedelta(minutes=rng.randint(0, 525600))
    return [full_name, email, signup_source, created_at.isoformat(sep=" ")]


def win1252_factory(rng: random.Random, index: int) -> Iterable[str]:
    names = ["Renée Faßbinder", "Jürgen Müller", "François L’Écuyer", "María-José"]
    full_name = rng.choice(names)
    email = f"cp1252_{index}@example.com"
    signup_source = "trade—show"  # em dash
    created_at = datetime(2024, 8, 1) + timedelta(minutes=rng.randint(0, 525600))
    return [full_name, email, signup_source, created_at.isoformat(sep=" ")]


def extra_columns_factory(rng: random.Random, index: int) -> Iterable[str]:
    row = list(base_row_factory(rng, index))
    row.append(f"unexpected-field-{index}")
    return row


def type_mismatch_factory(rng: random.Random, index: int) -> Iterable[str]:
    quantity = rng.randint(1, 10)
    order_total = f"${quantity * rng.uniform(10.0, 199.0):.2f}"
    return [index + 1, rng.randint(1, 5000), quantity, order_total]


def missing_quotes_writer(path: Path) -> None:
    content = (
        "full_name,email,signup_source,created_at\n"
        '"Anne Wong,email@example.com,referral,2024-05-05 10:10:00\n'
        'Carlos "Chaz" Ortiz,carlos@example.com,website,2024-05-05 11:00:00\n'
    )
    path.write_text(content, encoding="utf-8")


def build_profiles(args: argparse.Namespace) -> dict[str, Profile]:
    return {
        "base": Profile("base", DEFAULT_HEADERS, delimiter=args.delimiter, encoding=args.encoding, row_factory=base_row_factory),
        "semicolon": Profile("semicolon", DEFAULT_HEADERS, delimiter=";", row_factory=base_row_factory),
        "tab": Profile("tab", DEFAULT_HEADERS, delimiter="\t", row_factory=base_row_factory),
        "quoted_commas": Profile("quoted_commas", DEFAULT_HEADERS, row_factory=quoted_commas_factory),
        "multiline": Profile("multiline", [*DEFAULT_HEADERS, "address"], row_factory=multiline_factory),
        "latin1": Profile("latin1", DEFAULT_HEADERS, encoding="latin-1", row_factory=latin1_factory),
        "windows1252": Profile("windows1252", DEFAULT_HEADERS, encoding="cp1252", row_factory=win1252_factory),
        "extra_columns": Profile("extra_columns", DEFAULT_HEADERS, row_factory=extra_columns_factory),
        "type_mismatch": Profile(
            "type_mismatch",
            ["order_id", "customer_id", "quantity", "order_total"],
            row_factory=type_mismatch_factory,
        ),
        "missing_quotes": Profile(
            "missing_quotes",
            DEFAULT_HEADERS,
            uses_csv_writer=False,
            custom_writer=missing_quotes_writer,
        ),
    }


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Generate CSV datasets for PostgreSQL COPY tests")
    parser.add_argument("--rows", type=int, required=True, help="Number of rows to generate")
    parser.add_argument("--output", type=str, required=True, help="Output file path")
    parser.add_argument("--delimiter", type=str, default=",", help="Field delimiter (default: ',') for base profile")
    parser.add_argument("--encoding", type=str, default="utf-8", help="File encoding for base profile")
    parser.add_argument("--seed", type=int, default=42, help="Random seed for reproducibility")
    parser.add_argument(
        "--profile",
        type=str,
        default="base",
        choices=[
            "base",
            "semicolon",
            "tab",
            "quoted_commas",
            "multiline",
            "latin1",
            "windows1252",
            "extra_columns",
            "type_mismatch",
            "missing_quotes",
        ],
        help="Dataset profile to generate",
    )
    return parser


def ensure_parent_dir(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)


def generate_with_profile(profile: Profile, path: Path, rows: int, seed: int) -> None:
    ensure_parent_dir(path)
    rng = random.Random(seed)

    if not profile.uses_csv_writer:
        if profile.custom_writer is None:
            raise ValueError(f"Profile {profile.name} requires custom_writer")
        profile.custom_writer(path)
        return

    with path.open("w", encoding=profile.encoding, newline="") as handle:
        writer = csv.writer(handle, delimiter=profile.delimiter, quoting=csv.QUOTE_MINIMAL)
        writer.writerow(profile.headers)
        for index in range(rows):
            row = profile.row_factory(rng, index)
            writer.writerow(row)


def main() -> None:
    parser = build_parser()
    args = parser.parse_args()
    profiles = build_profiles(args)
    profile = profiles[args.profile]

    if profile.name == "base":
        profile.delimiter = args.delimiter
        profile.encoding = args.encoding

    generate_with_profile(profile, Path(args.output), rows=args.rows, seed=args.seed)


if __name__ == "__main__":
    main()