pgx copyfrom with WHERE condition

Question

I would like to use pgx (via sqlc) copyfrom feature with a COPY FROM WHERE condition as such:

(this condition was taken from a working rule on INSERT):

    WHERE (EXISTS ( SELECT *
                FROM asns
                WHERE merchantId IS NOT DISTINCT FROM NEW.merchantId
                AND return_provider IS NOT DISTINCT FROM NEW.return_provider
                AND barcode IS NOT DISTINCT FROM NEW.barcode
                AND carrier IS NOT DISTINCT FROM NEW.carrier
                AND tracking_number IS NOT DISTINCT FROM NEW.tracking_number
                AND customer_email IS NOT DISTINCT FROM NEW.customer_email
                AND order_id IS NOT DISTINCT FROM NEW.order_id
                AND order_name IS NOT DISTINCT FROM NEW.order_name
                AND order_number IS NOT DISTINCT FROM NEW.order_number
                AND return_line_item_id IS NOT DISTINCT FROM NEW.return_line_item_id
                AND rma IS NOT DISTINCT FROM NEW.rma
                AND sku IS NOT DISTINCT FROM NEW.sku)) DO INSTEAD NOTHING;

Is there a way to do this in pgx? Thank you.

Answer 1

Yes, there is a way of doing this. One can use

func (pgConn *PgConn) CopyFrom(ctx context.Context, r io.Reader, sql string) (CommandTag, error).

You can find the example of usage in test cases:

func TestConnCopyFrom(t *testing.T) {
    t.Parallel()

    pgConn, err := pgconn.Connect(context.Background(), os.Getenv("PGX_TEST_DATABASE"))
    require.NoError(t, err)
    defer closeConn(t, pgConn)

    if pgConn.ParameterStatus("crdb_version") != "" {
        t.Skip("Server does not fully support COPY FROM (https://www.cockroachlabs.com/docs/v20.2/copy-from.html)")
    }

    _, err = pgConn.Exec(context.Background(), `create temporary table foo(
        a int4,
        b varchar
    )`).ReadAll()
    require.NoError(t, err)

    srcBuf := &bytes.Buffer{}

    inputRows := [][][]byte{}
    for i := 0; i < 1000; i++ {
        a := strconv.Itoa(i)
        b := "foo " + a + " bar"
        inputRows = append(inputRows, [][]byte{[]byte(a), []byte(b)})
        _, err = srcBuf.Write([]byte(fmt.Sprintf("%s,\"%s\"\n", a, b)))
        require.NoError(t, err)
    }

    ct, err := pgConn.CopyFrom(context.Background(), srcBuf, "COPY foo FROM STDIN WITH (FORMAT csv)")
    require.NoError(t, err)
    assert.Equal(t, int64(len(inputRows)), ct.RowsAffected())

    result := pgConn.ExecParams(context.Background(), "select * from foo", nil, nil, nil, nil).Read()
    require.NoError(t, result.Err)

    assert.Equal(t, inputRows, result.Rows)

    ensureConnValid(t, pgConn)
}

Just extend COPY foo FROM STDIN ... with WHERE statement.