PostgreSQL使用COPY协议高效批量数据写入的实战指南

pq: got 86575 parameters but postgresql only supports 65535 parameters

import (
    "github.com/lib/pq"
    "gorm.io/gorm"
)

// batchcreate 批量创建消息接收者记录 - 使用 copy 协议
func (r *receiverrepo) batchcreate(ctx context.context, db *gorm.db, data []*define.wecommsgreceiver) (rowsaffected int64, err error) {
    db = r.withtrace(ctx, db)
    db = db.table(r.tablename())

    if len(data) == 0 {
        return 0, nil
    }

    // 过滤掉 nil 的数据
    validdata := make([]*define.wecommsgreceiver, 0, len(data))
    for _, item := range data {
        if item != nil {
            validdata = append(validdata, item)
        }
    }
    if len(validdata) == 0 {
        return 0, nil
    }

    // 获取底层 sql.db
    sqldb := db.db()
    
    // 开始事务
    tx, err := sqldb.begintx(ctx, nil)
    if err != nil {
        return 0, fmt.errorf("开始事务失败：%+v", err)
    }
    defer func() {
        if err != nil {
            tx.rollback()
        }
    }()

    // 创建 copy writer
    stmt, err := tx.prepare(pq.copyin(r.tablename(), "send_log_id", "user_id", "status", "created_at", "updated_at"))
    if err != nil {
        return 0, fmt.errorf("准备 copy 语句失败：%+v", err)
    }
    defer stmt.close()

    // 批量写入数据
    for _, item := range validdata {
        _, err = stmt.exec(item.sendlogid, item.userid, item.status, item.createdat, item.updatedat)
        if err != nil {
            return 0, fmt.errorf("写入数据失败：%+v", err)
        }
    }

    // 执行 copy
    _, err := stmt.exec()
    if err != nil {
        return 0, fmt.errorf("执行 copy 失败：%+v", err)
    }

    // 提交事务
    if err = tx.commit(); err != nil {
        return 0, fmt.errorf("提交事务失败：%+v", err)
    }
    
    rowsaffected = int64(len(validdata))
    return rowsaffected, nil
}

// 设置测试数据库
func setuptestdb() (*gorm.db, error) {
    ctx := context.background()
    postgres, err := infrastructure.dialpostgres(ctx, infrastructure.postgresconfig{
        host:     "host",
        port:     5432,
        username: "postgres",
        password: "xxxxx",
        database: "xxxxx",
    })
    if err != nil {
        return nil, err
    }

    return postgres, nil
}

func setuplogger() factory.logfactory {
    logger, _ := factory.newjsonfactory(factory.newlevel("info"), factory.newzapoption(factory.addcallerskip(0)))
    return logger
}

func testreceiverrepo_batchcreate(t *testing.t) {
    db, err := setuptestdb()
    require.noerror(t, err)
    defer db.close()

    // 创建日志工厂
    logger := setuplogger()

    // 创建 repository 实例
    repo := newreceiverrepository(db, logger)

    // 准备测试数据 - 20000 条记录，使用负的 send_log_id 避免污染数据
    testdata := make([]*define.wecommsgreceiver, 0, 20000)
    now := time.now()
    negativesendlogid := int64(-100000) // 使用负的 send_log_id

    for i := 0; i < 20000; i++ {
        testdata = append(testdata, &define.wecommsgreceiver{
            sendlogid: negativesendlogid,
            userid:    "test_user_" + fmt.sprint(i),
            status:    1,
            createdat: now,
            updatedat: now,
        })
    }

    ctx := context.background()

    // 执行批量插入
    rowsaffected, err := repo.batchcreate(ctx, db, testdata)

    // 验证结果
    assert.noerror(t, err)
    assert.equal(t, int64(20000), rowsaffected)

    // 验证数据是否正确插入
    var count int64
    query := "select count(*) from wecom_msg_receiver where send_log_id < 0 and send_log_id >= ?"
    err = db.raw(query, negativesendlogid).count(&count).error
    assert.noerror(t, err)
    assert.equal(t, int64(20000), count)

    // 清理测试数据
    deletequery := "delete from wecom_msg_receiver where send_log_id < 0 and send_log_id >= ?"
    result := db.exec(deletequery, negativesendlogid)
    assert.noerror(t, result.error)
    assert.equal(t, int64(20000), result.rowsaffected)

    // 验证清理是否成功
    err = db.raw(query, negativesendlogid).count(&count).error
    assert.noerror(t, err)
    assert.equal(t, int64(0), count)
}