copy any not bytes

kinda silly decoding values only to reencode, but getting around would mean lifting pgx code to send copy ourselves

flow/activities/flowable.go

@@ -453,7 +453,7 @@ func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
 				connectors.QRepSyncConnector.SyncQRepRecords,
 			)
 		case protos.TypeSystem_PG:
-			stream := model.NewRecordStream[[]byte](shared.FetchAndChannelSize)
+			stream := model.NewRecordStream[any](shared.FetchAndChannelSize)
 			err = replicateQRepPartition(ctx, a, config, i+1, numPartitions, p, runUUID,
 				stream, stream,
 				connectors.QRepPullPgConnector.PullPgQRepRecords,
@@ -776,7 +776,7 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
 			connectors.QRepSyncConnector.SyncQRepRecords)
 	case protos.TypeSystem_PG:
 		return replicateXminPartition(ctx, a, config, partition, runUUID,
-			model.NewRecordStream[[]byte](shared.FetchAndChannelSize),
+			model.NewRecordStream[any](shared.FetchAndChannelSize),
 			(*connpostgres.PostgresConnector).PullXminPgRecordStream,
 			connectors.QRepSyncPgConnector.SyncPgQRepRecords)
 	default:

flow/connectors/postgres/qrep_query_executor.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
-	"slices"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
@@ -432,10 +431,6 @@ func (qe *QRepQueryExecutor) mapRowToQRecord(
 func (qe *QRepQueryExecutor) mapRowToPgRecord(
 	row pgx.Rows,
 	_ []pgconn.FieldDescription,
-) ([][]byte, error) {
-	raw := row.RawValues()
-	for i, val := range raw {
-		raw[i] = slices.Clone(val)
-	}
-	return slices.Clone(raw), nil
+) ([]any, error) {
+	return row.Values()
 }

flow/model/pgrecord_copy_from_source.go

@@ -1,12 +1,8 @@
 package model
 
-import (
-	"github.com/PeerDB-io/peer-flow/shared"
-)
-
 type PgRecordCopyFromSource struct {
 	stream        *PgRecordStream
-	currentRecord [][]byte
+	currentRecord []any
 }
 
 func NewPgRecordCopyFromSource(
@@ -21,19 +17,11 @@ func NewPgRecordCopyFromSource(
 func (src *PgRecordCopyFromSource) Next() bool {
 	rec, ok := <-src.stream.Records
 	src.currentRecord = rec
-	return ok
+	return ok || src.Err() != nil
 }
 
 func (src *PgRecordCopyFromSource) Values() ([]interface{}, error) {
-	if err := src.Err(); err != nil {
-		return nil, err
-	}
-
-	values := make([]interface{}, len(src.currentRecord))
-	for i, val := range src.currentRecord {
-		values[i] = shared.UnsafeFastReadOnlyBytesToString(val)
-	}
-	return values, nil
+	return src.currentRecord, src.Err()
 }
 
 func (src *PgRecordCopyFromSource) Err() error {

flow/model/qrecord_copy_from_source.go

@@ -42,7 +42,7 @@ func NewQRecordCopyFromSource(
 func (src *QRecordCopyFromSource) Next() bool {
 	rec, ok := <-src.stream.Records
 	src.currentRecord = rec
-	return ok
+	return ok && src.Err() != nil
 }
 
 func (src *QRecordCopyFromSource) Values() ([]interface{}, error) {

flow/model/qrecord_stream.go

@@ -22,7 +22,7 @@ type RecordStream[T any] struct {
 
 type (
 	QRecordStream  = RecordStream[qvalue.QValue]
-	PgRecordStream = RecordStream[[]byte]
+	PgRecordStream = RecordStream[any]
 )
 
 type RecordsToStreamRequest[T Items] struct {