How to use CopyFrom with a sql.RawBytes or generic interfaces for Timestamp and other data types?

[kalensk]

Basically, I need a "generic" query since I don't want my application to be tied to the underlying table structure (its complicated). However scanning into an interface or rawbytes fails when running CopyFrom() due to not being able to encode timestamp data type. Any ideas or suggestions @jackc ?

type Result struct {
	Query   string
	Rows    [][]any
	Columns []string
}


func (d *SqlConnection) QueryForRowsAndColumns(query string) (*Result, error) {
	// query is essentially SELECT * FROM $table;
        rowResult, err := d.SqlDb().Query(query)
	if err != nil {
		return nil, fmt.Errorf("querying: %w", err)
	}
	defer func() {
		if cErr := rowResult.Close(); cErr != nil {
			log.Print(cErr)
			err = errors.Join(err, cErr)
		}
	}()

	columns, err := rowResult.Columns()
	if err != nil {
		return nil, err
	}

	rowValues := make([]any, len(columns))
	for i, _ := range columns {
		rowValues[i] = new(sql.RawBytes)
	}

	var rows [][]any
	for rowResult.Next() {
		err = rowResult.Scan(rowValues...)
		if err != nil {
			return nil, fmt.Errorf("scanning: %w", err)
		}

		rows = append(rows, rowValues)
	}

	err = rowResult.Err()
	if err != nil {
		return nil, err
	}

	return &Result{
		Query:   query,
		Rows:    rows,
		Columns: columns,
	}, nil
}

I do some business logic on the rows, and later on in my code we want to do a CopyFrom into another database such as:

_, err := db.PgxConn().CopyFrom(
		context.Background(),
		pgx.Identifier{table.Schema, table.Name},
		input.Columns,
		pgx.CopyFromRows(input.Rows),
	)
	if err != nil {
		return nil, err
	}

which failed with:

│ERROR: COPY from stdin failed: unable to encode &sql.RawBytes{0x32, 0x30, 0x32, 0x35, 0x2d, 0x30, 0x32, 0x2d, 0x32, 0x37, 0x54, 0x31, 0x33, 0x3a, 0x35, 0x35, 0x3a, 0x33, 0x33, 0x2e, 0x30, 0x34, 0x31, 0x34, 0x31, 0x36, 0x5a} into binary format │
│                                                                                        for timestamp (OID 1114): cannot find encode plan (SQLSTATE 57014)    

Also tried something like:

	// query is essentially SELECT * FROM $table;

        var rows [][]any
	for rowResult.Next() {
		rowValues := make([]interface{}, len(columns))
		for i := range columns {
			rowValues[i] = new(interface{})
		}

		err = rowResult.Scan(rowValues...)
		if err != nil {
			return nil, fmt.Errorf("QueryForRowsAndColumns: %w", err)
		}

		var row []any
		for _, rowValue := range rowValues {
			value := fmt.Sprintf("%v", *rowValue.(*interface{}))
			row = append(row, fmt.Sprintf("%v", value))
		}

		rows = append(rows, row)
	}

which failed with:

COPY from stdin failed: unable to encode "2025-02-26 18:40:29.702997 +0000 UTC" into binary format for timestamp (OID 1114): cannot find encode plan 

[jackc]

There isn't a way to send raw bytes to CopyFrom() (or as a query arg for that matter). pgx's scan path allows using pgtype.UndecodedBytes to read the raw value from over the wire. I suppose the inverse could be built into the encode path.

But perhaps it would be simpler to change your query path. If you used the pgx interface and called Values() you would get appropriately typed values that should be sendable via CopyFrom().

[kalensk]

Thank you @jackc for the reply. After much more headbanging and subtle bugs I am trying a new approach which I "think" might be better, but I could really use your assistance for part of it. Again the goal is to be "generic" as I don't want my application to be tied to the underlying table structure (its complicated).

Basically, I am doing pgxConn.Query("SELECT * FROM myTable;") and gathering the rows as following. I am definitely open to other suggestions on collecting the rows to better help me serialize them to csv for later ingestion by COPY FROM in postgres.

var rows [][]any
	for rowResult.Next() {
		rowValues, err := rowResult.Values()
		if err != nil {
			return nil, err
		}

		rows = append(rows, rowValues)
	}

Later on as part of my application I need to write out a .csv file of these rows such that it can later be ingested via postgres's COPY FROM via psql. That is, it should be possible to run COPY myNewTable FROM '/tmp/output.csv' DELIMITER ',' NULL 'NULL' CSV HEADER; with the written out csv file from the rows collected above.

However, to write the csv it needs to write the fields correctly for COPY FROM to work correctly. For example, for TIMESTAMP it should be 2025-03-05 16:32:01.803465 and NOT 2025-03-05 16:32:01.803465 +0000 UTC. And for bytea as \x116114 ... and NOT [116 114 ..., etc.

The issue is I am having trouble finding a generic way to correctly write out the rows which have type [][]any such that they get printed correctly as explained in the previous paragraph. Naively, my first approach is the following:

for _, row := range q.Rows {
	var line []string
	for _, field := range row {
		fieldAsString := fmt.Sprint(field)
		line = append(line, fieldAsString)
	}

	err := csvWriter.Write(line)
	if err != nil {
		return "", fmt.Errorf("writing csv line %q: %w", line, err)
	}
}

Maybe I can save pgxConn.TypeMap() and fieldDescriptions for use that later on when writing the csv to convert it to the correct type to print out the csv rows?!? How might that look exactly? Roughly something like typeMap.TypeForOID(fieldDescription.DataTypeOID).PlanEncode().Encode()...?!? Hmm typeMap.TypeForOID() doesn't seem to work for string, but does for other more complex types...

Ugh! It looks like *pgxpool.Pool does not have a way to get the TypeMap()? Oh I suppose I'd have to do the following and save the TypeMap when the pool is first made. So something like:

pgxConfig, err := pgxpool.ParseConfig(myConnectionString)
	if err != nil {
		return nil, fmt.Errorf("unable to parse config: %v", err)
	}

	var typeMap *pgtype.Map
	pgxConfig.BeforeAcquire = func(ctx context.Context, conn *pgx.Conn) bool {
		typeMap = conn.TypeMap() // save typeMap
		return true
	}

	conn, err := pgxpool.NewWithConfig(context.Background(), pgxConfig)
	if err != nil {
		return nil, err
	}

The following seems to work I think.

	err = csvWriter.Write(headers)
	if err != nil {
		return "", fmt.Errorf("writing csv columns %q: %w", p.FieldDescriptions, err)
	}

	for _, row := range p.Rows {
		var line []string
		for i, field := range row {
			encodedField, err := p.encodeField(field, p.FieldDescriptions[i])
			if err != nil {
				return "", err
			}

			line = append(line, encodedField)
		}

		err := csvWriter.Write(line)
		if err != nil {
			return "", fmt.Errorf("writing csv line %q: %w", line, err)
		}
	}

// .........

func (p PgxResult) encodeField(field any, fieldDescription pgconn.FieldDescription) (string, error) {
	encodedField, err := p.TypeMap.Encode(fieldDescription.DataTypeOID, pgtype.TextFormatCode, field, nil)
	if err != nil {
		return "", err
	}

	if encodedField == nil {
		return "NULL", nil
	}

	return string(encodedField), nil
}

Any ideas?