feat: add casting for special postgres types
This commit is contained in:
@@ -10,6 +10,7 @@ import (
|
|||||||
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
|
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
|
||||||
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db"
|
"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/db"
|
||||||
"github.com/jackc/pgx/v5"
|
"github.com/jackc/pgx/v5"
|
||||||
|
"github.com/jackc/pgx/v5/pgtype"
|
||||||
"github.com/jackc/pgx/v5/pgxpool"
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
)
|
)
|
||||||
@@ -47,18 +48,18 @@ func main() {
|
|||||||
table := "migration_test"
|
table := "migration_test"
|
||||||
colNames := []string{"id", "nombre_producto", "descripcion", "stock", "precio", "es_activo", "fecha_creacion", "ultima_actualizacion", "configuracion_json", "etiquetas", "binario_test", "ip_servidor", "rango_prueba"}
|
colNames := []string{"id", "nombre_producto", "descripcion", "stock", "precio", "es_activo", "fecha_creacion", "ultima_actualizacion", "configuracion_json", "etiquetas", "binario_test", "ip_servidor", "rango_prueba"}
|
||||||
|
|
||||||
rowValues, err := extractData(ctxSource, sourcePool, schema, table, colNames, 2)
|
rowValues, err := extractData(ctxSource, sourcePool, schema, table, colNames, 10000)
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatal("Unexpected error when extracting data", err)
|
log.Fatal("Unexpected error when extracting data", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
for index, row := range rowValues {
|
// for index, row := range rowValues {
|
||||||
log.Debugf("Values for row %d", index+1)
|
// log.Debugf("Values for row %d", index+1)
|
||||||
for i, v := range row {
|
// for i, v := range row {
|
||||||
log.Debugf("%s: %v", colNames[i], v)
|
// log.Debugf("%s: %v", colNames[i], v)
|
||||||
}
|
// }
|
||||||
}
|
// }
|
||||||
|
|
||||||
insertedRows, err := insertData(ctxTarget, targetPool, schema, table, colNames, rowValues)
|
insertedRows, err := insertData(ctxTarget, targetPool, schema, table, colNames, rowValues)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -106,28 +107,84 @@ func extractData(ctx context.Context, sourcePool *pgxpool.Pool, schema string, t
|
|||||||
defer rows.Close()
|
defer rows.Close()
|
||||||
|
|
||||||
cols := rows.FieldDescriptions()
|
cols := rows.FieldDescriptions()
|
||||||
for _, c := range cols {
|
oids := make([]uint32, len(cols))
|
||||||
log.Debugf("Column: %+v", c)
|
for i, c := range cols {
|
||||||
|
oids[i] = c.DataTypeOID
|
||||||
}
|
}
|
||||||
|
|
||||||
rowValues := make([][]any, 0, limit)
|
rowValues := make([][]any, 0, limit)
|
||||||
|
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
values := make([]any, len(cols))
|
values, _ := rows.Values()
|
||||||
valuePtrs := make([]any, len(cols))
|
|
||||||
|
|
||||||
for i := range values {
|
for i, v := range values {
|
||||||
valuePtrs[i] = &values[i]
|
values[i] = castValueByOID(v, oids[i])
|
||||||
}
|
}
|
||||||
|
|
||||||
rows.Scan(valuePtrs...)
|
|
||||||
|
|
||||||
rowValues = append(rowValues, values)
|
rowValues = append(rowValues, values)
|
||||||
}
|
}
|
||||||
|
|
||||||
return rowValues, nil
|
return rowValues, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func castValueByOID(val any, oid uint32) any {
|
||||||
|
if val == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
switch oid {
|
||||||
|
case 3904:
|
||||||
|
if r, ok := val.(pgtype.Range[any]); ok {
|
||||||
|
newRange := pgtype.Range[int32]{
|
||||||
|
LowerType: r.LowerType,
|
||||||
|
UpperType: r.UpperType,
|
||||||
|
Valid: r.Valid,
|
||||||
|
}
|
||||||
|
if r.Lower != nil {
|
||||||
|
newRange.Lower = anyToInt32(r.Lower)
|
||||||
|
}
|
||||||
|
if r.Upper != nil {
|
||||||
|
newRange.Upper = anyToInt32(r.Upper)
|
||||||
|
}
|
||||||
|
return newRange
|
||||||
|
}
|
||||||
|
|
||||||
|
case 3906:
|
||||||
|
if r, ok := val.(pgtype.Range[any]); ok {
|
||||||
|
newRange := pgtype.Range[pgtype.Numeric]{
|
||||||
|
LowerType: r.LowerType,
|
||||||
|
UpperType: r.UpperType,
|
||||||
|
Valid: r.Valid,
|
||||||
|
}
|
||||||
|
if r.Lower != nil {
|
||||||
|
newRange.Lower = r.Lower.(pgtype.Numeric)
|
||||||
|
}
|
||||||
|
if r.Upper != nil {
|
||||||
|
newRange.Upper = r.Upper.(pgtype.Numeric)
|
||||||
|
}
|
||||||
|
return newRange
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return val
|
||||||
|
}
|
||||||
|
|
||||||
|
func anyToInt32(v any) int32 {
|
||||||
|
switch t := v.(type) {
|
||||||
|
case int32:
|
||||||
|
return t
|
||||||
|
case int64:
|
||||||
|
return int32(t)
|
||||||
|
case int:
|
||||||
|
return int32(t)
|
||||||
|
case float64:
|
||||||
|
return int32(t)
|
||||||
|
default:
|
||||||
|
log.Warnf("Valor inesperado en rango: %T", v)
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func insertData(ctx context.Context, targetPool *pgxpool.Pool, schema string, table string, colNames []string, rowValues [][]any) (int64, error) {
|
func insertData(ctx context.Context, targetPool *pgxpool.Pool, schema string, table string, colNames []string, rowValues [][]any) (int64, error) {
|
||||||
identifier := pgx.Identifier{schema, table}
|
identifier := pgx.Identifier{schema, table}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user