Go pgx batch insert performance

Import 1 juta baris produk dari CSV pakai INSERT loop di Go? Siap-siap nunggu 30 menit. CopyFrom pgx pakai protocol COPY Postgres binary — selesai dalam menit. Snippet ini compare 3 approach + handle ON CONFLICT via temp table pattern.

Kode

package importer

import (
	"context"
	"fmt"
	"time"

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

type Produk struct {
	SKU       string
	Nama      string
	Harga     int
	Stok      int
	UpdatedAt time.Time
}

// CARA 1 — SLOW: insert per row
func InsertOneByOne(ctx context.Context, db *pgxpool.Pool, items []Produk) error {
	for _, p := range items {
		_, err := db.Exec(ctx,
			`INSERT INTO produk (sku, nama, harga, stok, updated_at)
			 VALUES ($1, $2, $3, $4, $5)`,
			p.SKU, p.Nama, p.Harga, p.Stok, p.UpdatedAt,
		)
		if err != nil {
			return fmt.Errorf("insert %s: %w", p.SKU, err)
		}
	}
	return nil
}

// CARA 2 — Batch dengan pgx.Batch (good)
func InsertBatch(ctx context.Context, db *pgxpool.Pool, items []Produk) error {
	batch := &pgx.Batch{}
	for _, p := range items {
		batch.Queue(
			`INSERT INTO produk (sku, nama, harga, stok, updated_at)
			 VALUES ($1, $2, $3, $4, $5)`,
			p.SKU, p.Nama, p.Harga, p.Stok, p.UpdatedAt,
		)
	}

	br := db.SendBatch(ctx, batch)
	defer br.Close()

	for range items {
		if _, err := br.Exec(); err != nil {
			return fmt.Errorf("batch exec: %w", err)
		}
	}
	return nil
}

// CARA 3 — CopyFrom (TERCEPAT)
func InsertCopy(ctx context.Context, db *pgxpool.Pool, items []Produk) (int64, error) {
	rows := make([][]any, len(items))
	for i, p := range items {
		rows[i] = []any{p.SKU, p.Nama, p.Harga, p.Stok, p.UpdatedAt}
	}

	copied, err := db.CopyFrom(
		ctx,
		pgx.Identifier{"produk"},
		[]string{"sku", "nama", "harga", "stok", "updated_at"},
		pgx.CopyFromRows(rows),
	)
	if err != nil {
		return 0, fmt.Errorf("copy from: %w", err)
	}
	return copied, nil
}

// CARA 4 — CopyFrom dengan ON CONFLICT (upsert) via temp table
func UpsertCopy(ctx context.Context, db *pgxpool.Pool, items []Produk) (int64, error) {
	tx, err := db.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback(ctx)

	// 1. Bikin temp table sesuai schema target
	_, err = tx.Exec(ctx, `
		CREATE TEMP TABLE produk_staging (LIKE produk INCLUDING DEFAULTS)
		ON COMMIT DROP
	`)
	if err != nil {
		return 0, fmt.Errorf("create temp: %w", err)
	}

	// 2. CopyFrom ke temp (cepat)
	rows := make([][]any, len(items))
	for i, p := range items {
		rows[i] = []any{p.SKU, p.Nama, p.Harga, p.Stok, p.UpdatedAt}
	}

	_, err = tx.CopyFrom(
		ctx,
		pgx.Identifier{"produk_staging"},
		[]string{"sku", "nama", "harga", "stok", "updated_at"},
		pgx.CopyFromRows(rows),
	)
	if err != nil {
		return 0, fmt.Errorf("copy staging: %w", err)
	}

	// 3. Merge ke target dengan ON CONFLICT
	tag, err := tx.Exec(ctx, `
		INSERT INTO produk (sku, nama, harga, stok, updated_at)
		SELECT sku, nama, harga, stok, updated_at FROM produk_staging
		ON CONFLICT (sku) DO UPDATE SET
			nama       = EXCLUDED.nama,
			harga      = EXCLUDED.harga,
			stok       = EXCLUDED.stok,
			updated_at = EXCLUDED.updated_at
	`)
	if err != nil {
		return 0, fmt.Errorf("merge: %w", err)
	}

	if err := tx.Commit(ctx); err != nil {
		return 0, err
	}
	return tag.RowsAffected(), nil
}

Pemakaian

package main

import (
	"context"
	"encoding/csv"
	"log"
	"os"
	"strconv"
	"time"

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

func main() {
	ctx := context.Background()
	db, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	// Baca CSV produk
	f, err := os.Open("produk_tokopedia.csv")
	if err != nil {
		log.Fatal(err)
	}
	defer f.Close()

	reader := csv.NewReader(f)
	_, _ = reader.Read() // skip header

	var items []Produk
	for {
		row, err := reader.Read()
		if err != nil {
			break
		}
		harga, _ := strconv.Atoi(row[2])
		stok, _ := strconv.Atoi(row[3])
		items = append(items, Produk{
			SKU:       row[0],
			Nama:      row[1],
			Harga:     harga,
			Stok:      stok,
			UpdatedAt: time.Now(),
		})
	}

	log.Printf("Total produk: %d", len(items))

	start := time.Now()
	count, err := UpsertCopy(ctx, db, items)
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("Upsert %d rows dalam %s", count, time.Since(start))
}

Benchmark 100rb baris produk:
- InsertOneByOne:  187 detik
- InsertBatch:      14 detik
- InsertCopy:        2 detik
- UpsertCopy:        3 detik (dengan ON CONFLICT)

Kapan dipakai

• Import data dari CSV / Excel manual.
• Sync produk dari supplier API ke katalog internal.
• ETL pipeline harian dari source eksternal.
• Initial data seed saat migrasi sistem.
• Restore parsial dari backup logical.

Catatan

• CopyFromRows convenient tapi materialize semua rows ke memory dulu. Untuk dataset besar, implementasi pgx.CopyFromSource interface dengan iterator streaming.
• No ON CONFLICT di CopyFrom — workaround temp table seperti contoh 4. Tetap dramatis lebih cepat dari INSERT loop.
• Chunk besar — kalau insert > 1 juta row, split jadi batch 50k-100k per CopyFrom. Hindari one giant transaction supaya commit gak lama.
• WAL bloat — bulk insert generate banyak WAL. Untuk one-time massive import, pertimbangkan wal_level=minimal (di non-replication setup) dan disable index dulu, rebuild setelah.
• Type mapping — time.Time ke timestamptz, int ke int4. Pastikan match kolom. pgx kasih error jelas kalau mismatch.

Untuk dataset 100GB+, pgx CopyFrom tetap kalah dari psql \copy dengan file. Generate CSV dari Go ke disk, lalu shell out ke psql — kadang yang paling lama lifetime tetap-paling-cepat.