feat: add type transformations (uuid, wkb, datetime)

Apply row-level transformations for special mssql types: uniqueidentifier, geometry/geography (WKB to EWKB) and datetime (UTC).

cmd/go_migrate/build-extract-query.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+)
+
+func buildExtractQueryMssql(job MigrationJob, columns []ColumnType) string {
+	var sbColumns strings.Builder
+
+	if len(columns) == 0 {
+		sbColumns.WriteString("*")
+	} else {
+		for i, col := range columns {
+			sbColumns.WriteString("[")
+			sbColumns.WriteString(col.name)
+			sbColumns.WriteString("]")
+
+			if col.unifiedType == "GEOMETRY" {
+				sbColumns.WriteString(".STAsBinary() AS [")
+				sbColumns.WriteString(col.name)
+				sbColumns.WriteString("]")
+			}
+
+			if i < len(columns)-1 {
+				sbColumns.WriteString(", ")
+			}
+		}
+	}
+
+	return fmt.Sprintf(`SELECT %s FROM [%s].[%s] ORDER BY [%s] ASC`, sbColumns.String(), job.Schema, job.Table, job.PrimaryKey)
+}
+
+func buildExtractQueryPostgres(job MigrationJob, columns []ColumnType) string {
+	var sbColumns strings.Builder
+
+	if len(columns) == 0 {
+		sbColumns.WriteString("*")
+	} else {
+		for i, col := range columns {
+			if col.unifiedType == "GEOMETRY" {
+				sbColumns.WriteString(`ST_AsEWKB("`)
+				sbColumns.WriteString(col.name)
+				sbColumns.WriteString(`") AS "`)
+				sbColumns.WriteString(col.name)
+				sbColumns.WriteString(`"`)
+			} else {
+				sbColumns.WriteString(`"`)
+				sbColumns.WriteString(col.name)
+				sbColumns.WriteString(`"`)
+			}
+
+			if i < len(columns)-1 {
+				sbColumns.WriteString(", ")
+			}
+		}
+	}
+
+	return fmt.Sprintf(`SELECT %s FROM "%s"."%s"  ORDER BY "%s" ASC`, sbColumns.String(), job.Schema, job.Table, job.PrimaryKey)
+}

cmd/go_migrate/colum-type.go

@@ -0,0 +1,44 @@
+package main
+
+type ColumnType struct {
+	name string
+
+	hasMaxLength      bool
+	hasPrecisionScale bool
+
+	userType    string
+	systemType  string
+	unifiedType string
+	nullable    bool
+	maxLength   int64
+	precision   int64
+	scale       int64
+}
+
+func (c *ColumnType) Name() string {
+	return c.name
+}
+
+func (c *ColumnType) UserType() string {
+	return c.userType
+}
+
+func (c *ColumnType) SystemType() string {
+	return c.systemType
+}
+
+func (c *ColumnType) Length() (length int64, ok bool) {
+	return c.maxLength, c.hasMaxLength
+}
+
+func (c *ColumnType) DecimalSize() (precision, scale int64, ok bool) {
+	return c.precision, c.scale, c.hasPrecisionScale
+}
+
+func (c *ColumnType) Nullable() bool {
+	return c.nullable
+}
+
+func (c *ColumnType) Type() string {
+	return c.unifiedType
+}

cmd/go_migrate/connect.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"git.ksdemosapps.com/kylesoda/go-migrate/internal/app/config"
+	"github.com/jackc/pgx/v5/pgxpool"
+	_ "github.com/microsoft/go-mssqldb"
+	log "github.com/sirupsen/logrus"
+)
+
+func connectToSqlServer() (*sql.DB, error) {
+	db, err := sql.Open("sqlserver", config.App.SourceDbUrl)
+	if err != nil {
+		return nil, fmt.Errorf("Unable to connect to sqlserver: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	if err := db.PingContext(ctx); err != nil {
+		return nil, fmt.Errorf("Unable to ping sqlserver: %w", err)
+	}
+
+	return db, nil
+}
+
+func connectToPostgres() (*pgxpool.Pool, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	pool, err := pgxpool.New(ctx, config.App.TargetDbUrl)
+	if err != nil {
+		return nil, fmt.Errorf("Unable to connect to postgres: %w", err)
+	}
+
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("Unable to ping postgres: %w", err)
+	}
+
+	return pool, nil
+}
+
+func connectToDatabases() (*sql.DB, *pgxpool.Pool, error) {
+	var sourceDbErr, targetDbErr error
+	var sourceDb *sql.DB
+	var targetDb *pgxpool.Pool
+	var wg sync.WaitGroup
+
+	wg.Go(func() {
+		sourceDb, sourceDbErr = connectToSqlServer()
+		if sourceDbErr != nil {
+			log.Error("Unable to connect to source db: ", sourceDbErr)
+		}
+	})
+
+	wg.Go(func() {
+		targetDb, targetDbErr = connectToPostgres()
+		if targetDbErr != nil {
+			log.Error("Unable to connect to target db: ", targetDbErr)
+		}
+	})
+
+	wg.Wait()
+
+	if sourceDbErr != nil || targetDbErr != nil {
+		return nil, nil, errors.New("Unable to connect to databases")
+	}
+
+	return sourceDb, targetDb, nil
+}

cmd/go_migrate/extractor.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	_ "github.com/microsoft/go-mssqldb"
+	log "github.com/sirupsen/logrus"
+)
+
+type UnknownRowValues []any
+
+func extractFromMssql(ctx context.Context, job MigrationJob, columns []ColumnType, chunkSize int, db *sql.DB, out chan<- []UnknownRowValues) error {
+	query := buildExtractQueryMssql(job, columns)
+	log.Debug("Query used to extract data from mssql: ", query)
+
+	rows, err := db.QueryContext(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	rowsChunk := make([]UnknownRowValues, 0, chunkSize)
+
+	for rows.Next() {
+		values := make([]any, len(columns))
+		scanArgs := make([]any, len(columns))
+
+		for i := range values {
+			scanArgs[i] = &values[i]
+		}
+
+		if err := rows.Scan(scanArgs...); err != nil {
+			return err
+		}
+
+		rowsChunk = append(rowsChunk, values)
+
+		if len(rowsChunk) >= chunkSize {
+			out <- rowsChunk
+			rowsChunk = make([]UnknownRowValues, 0, chunkSize)
+			log.Infof("Chunk send... %+v", job)
+		}
+	}
+
+	if len(rowsChunk) > 0 {
+		out <- rowsChunk
+		log.Infof("Chunk send... %+v", job)
+	}
+
+	return nil
+}
+
+func extractFromPostgres(ctx context.Context, job MigrationJob, columns []ColumnType, chunkSize int, db *pgxpool.Pool, out chan<- []UnknownRowValues) error {
+	query := buildExtractQueryPostgres(job, columns)
+	log.Debug("Query used to extract data from postgres: ", query)
+
+	rows, err := db.Query(ctx, query)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	rowsChunk := make([]UnknownRowValues, 0, chunkSize)
+
+	for rows.Next() {
+		values, err := rows.Values()
+		if err != nil {
+			return err
+		}
+
+		rowsChunk = append(rowsChunk, values)
+
+		if len(rowsChunk) >= chunkSize {
+			out <- rowsChunk
+			rowsChunk = make([]UnknownRowValues, 0, chunkSize)
+			log.Infof("Chunk send... %+v", job)
+		}
+	}
+
+	if len(rowsChunk) > 0 {
+		out <- rowsChunk
+		log.Infof("Chunk send... %+v", job)
+	}
+
+	return nil
+}

cmd/go_migrate/inspect-columns.go

@@ -0,0 +1,279 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+	_ "github.com/microsoft/go-mssqldb"
+	log "github.com/sirupsen/logrus"
+)
+
+func GetUnifiedType(systemType string) string {
+	systemType = strings.ToLower(systemType)
+
+	if systemType == "varchar" || systemType == "char" || systemType == "nvarchar" || systemType == "nchar" || systemType == "text" || systemType == "ntext" {
+		return "STRING"
+	}
+
+	if systemType == "int" || systemType == "int4" || systemType == "integer" || systemType == "smallint" || systemType == "int2" || systemType == "bigint" || systemType == "int8" || systemType == "tinyint" {
+		return "INTEGER"
+	}
+
+	if systemType == "decimal" || systemType == "numeric" {
+		return "DECIMAL"
+	}
+
+	if systemType == "float" || systemType == "real" || systemType == "double precision" {
+		return "FLOAT"
+	}
+
+	if systemType == "bit" || systemType == "boolean" {
+		return "BOOLEAN"
+	}
+
+	if systemType == "date" {
+		return "DATE"
+	}
+	if systemType == "time" || systemType == "time without time zone" {
+		return "TIME"
+	}
+	if systemType == "datetime" || systemType == "datetime2" || systemType == "timestamp" || systemType == "timestamptz" || systemType == "timestamp with time zone" {
+		return "TIMESTAMP"
+	}
+
+	if systemType == "binary" || systemType == "varbinary" || systemType == "image" || systemType == "bytea" {
+		return "BINARY"
+	}
+
+	if systemType == "uniqueidentifier" || systemType == "uuid" {
+		return "UUID"
+	}
+
+	if systemType == "json" {
+		return "JSON"
+	}
+
+	if systemType == "geometry" || systemType == "geography" {
+		return "GEOMETRY"
+	}
+
+	return strings.ToUpper(systemType)
+}
+
+func MapPostgresColumn(column ColumnType, maxLength *int64, precision *int64, scale *int64) ColumnType {
+	stringTypes := map[string]bool{
+		"varchar": true, "char": true, "character": true, "text": true, "character varying": true,
+	}
+
+	decimalTypes := map[string]bool{
+		"decimal": true, "numeric": true,
+	}
+
+	if stringTypes[column.systemType] {
+		if maxLength != nil {
+			column.maxLength = *maxLength
+			column.hasMaxLength = true
+		} else {
+			column.maxLength = -1
+			column.hasMaxLength = false
+		}
+		column.hasPrecisionScale = false
+		column.precision = -1
+		column.scale = -1
+	} else if decimalTypes[column.systemType] {
+		column.hasMaxLength = false
+		column.maxLength = -1
+		if precision != nil && scale != nil {
+			column.precision = *precision
+			column.scale = *scale
+			column.hasPrecisionScale = true
+		} else {
+			column.precision = -1
+			column.scale = -1
+			column.hasPrecisionScale = false
+		}
+	} else {
+		column.hasMaxLength = false
+		column.maxLength = -1
+		column.hasPrecisionScale = false
+		column.precision = -1
+		column.scale = -1
+	}
+
+	column.unifiedType = GetUnifiedType(column.systemType)
+
+	return column
+}
+
+func GetColumnTypesPostgres(db *pgxpool.Pool, migrationJob MigrationJob) ([]ColumnType, error) {
+	query := `
+SELECT 
+  c.column_name AS name,
+  c.data_type AS user_type,
+  c.udt_name AS system_type,
+  (CASE WHEN c.is_nullable = 'YES' THEN TRUE ELSE FALSE END) AS nullable,
+  c.character_maximum_length AS max_length,
+  c.numeric_precision AS precision,
+  c.numeric_scale AS scale
+FROM information_schema.columns c
+WHERE c.table_schema = $1 AND c.table_name = $2
+ORDER BY c.ordinal_position;
+`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	rows, err := db.Query(ctx, query, migrationJob.Schema, migrationJob.Table)
+	if err != nil {
+		return nil, fmt.Errorf("Error querying column types: %w", err)
+	}
+	defer rows.Close()
+
+	var colTypes []ColumnType
+
+	for rows.Next() {
+		var column ColumnType
+		var scanMaxLength *int64
+		var scanPrecision *int64
+		var scanScale *int64
+
+		if err := rows.Scan(
+			&column.name,
+			&column.userType,
+			&column.systemType,
+			&column.nullable,
+			&scanMaxLength,
+			&scanPrecision,
+			&scanScale,
+		); err != nil {
+			return nil, fmt.Errorf("Error scanning column type results: %w", err)
+		}
+
+		colTypes = append(colTypes, MapPostgresColumn(column, scanMaxLength, scanPrecision, scanScale))
+	}
+
+	return colTypes, nil
+}
+
+func MapMssqlColumn(column ColumnType) ColumnType {
+	stringTypes := map[string]bool{
+		"varchar": true, "char": true, "nvarchar": true, "nchar": true, "text": true, "ntext": true,
+	}
+
+	decimalTypes := map[string]bool{
+		"decimal": true, "numeric": true,
+	}
+
+	if stringTypes[column.systemType] {
+		column.hasMaxLength = true
+		if column.systemType == "nvarchar" || column.systemType == "nchar" {
+			if column.maxLength > 0 {
+				column.maxLength = column.maxLength / 2
+			}
+		}
+		column.hasPrecisionScale = false
+		column.precision = -1
+		column.scale = -1
+	} else if decimalTypes[column.systemType] {
+		column.hasMaxLength = false
+		column.maxLength = -1
+		column.hasPrecisionScale = true
+	} else {
+		column.hasMaxLength = false
+		column.maxLength = -1
+		column.hasPrecisionScale = false
+		column.precision = -1
+		column.scale = -1
+	}
+
+	column.unifiedType = GetUnifiedType(column.systemType)
+
+	return column
+}
+
+func GetColumnTypesMssql(db *sql.DB, migrationJob MigrationJob) ([]ColumnType, error) {
+	query := `
+SELECT 
+	c.name AS name,
+	t.name AS user_type,
+	CASE WHEN t.is_user_defined = 0 THEN t.name ELSE bt.name END AS system_type,
+	c.is_nullable AS nullable,
+	c.max_length AS max_length,
+	c.precision AS precision,
+	c.scale AS scale
+FROM sys.columns c
+JOIN sys.types t ON c.user_type_id = t.user_type_id
+LEFT JOIN sys.types bt ON t.is_user_defined = 1 AND bt.user_type_id = t.system_type_id
+JOIN sys.tables st ON c.object_id = st.object_id
+JOIN sys.schemas s ON st.schema_id = s.schema_id
+WHERE s.name = @schema AND st.name = @table
+ORDER BY c.column_id;
+`
+
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	rows, err := db.QueryContext(ctx, query, sql.Named("schema", migrationJob.Schema), sql.Named("table", migrationJob.Table))
+	if err != nil {
+		return nil, fmt.Errorf("Error querying column types: %w", err)
+	}
+	defer rows.Close()
+
+	var colTypes []ColumnType
+
+	for rows.Next() {
+		var column ColumnType
+
+		if err := rows.Scan(
+			&column.name,
+			&column.userType,
+			&column.systemType,
+			&column.nullable,
+			&column.maxLength,
+			&column.precision,
+			&column.scale,
+		); err != nil {
+			return nil, fmt.Errorf("Error scanning column type results: %W", err)
+		}
+
+		colTypes = append(colTypes, MapMssqlColumn(column))
+	}
+
+	return colTypes, nil
+}
+
+func GetColumnTypes(sourceDb *sql.DB, targetDb *pgxpool.Pool, migrationJob MigrationJob) ([]ColumnType, []ColumnType, error) {
+	var sourceDbErr error
+	var targetDbErr error
+	var sourceColTypes []ColumnType
+	var targetColTypes []ColumnType
+	var wg sync.WaitGroup
+
+	wg.Go(func() {
+		sourceColTypes, sourceDbErr = GetColumnTypesMssql(sourceDb, migrationJob)
+		if sourceDbErr != nil {
+			log.Error("Error (sourceDb): ", sourceDbErr)
+		}
+	})
+
+	wg.Go(func() {
+		targetColTypes, targetDbErr = GetColumnTypesPostgres(targetDb, migrationJob)
+		if targetDbErr != nil {
+			log.Error("Error (targetDb): ", targetDbErr)
+		}
+	})
+
+	wg.Wait()
+
+	if sourceDbErr != nil || targetDbErr != nil {
+		return nil, nil, errors.New("Error querying column types")
+	}
+
+	return sourceColTypes, targetColTypes, nil
+}

cmd/go_migrate/main.go

@@ -4,9 +4,43 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+type MigrationJob struct {
+	Schema     string
+	Table      string
+	PrimaryKey string
+}
+
+var migrationJobs []MigrationJob = []MigrationJob{
+	{
+		Schema:     "demo",
+		Table:      "users",
+		PrimaryKey: "id",
+	},
+}
+
+const (
+	NumExtractors int = 2
+	ChunkSize     int = 20
+	QueueSize     int = 10
+)
+
 func main() {
 	configureLog()
 	log.Info("Starting migration...")
+	// log.Debugf("Migration jobs: %+v", migrationJobs)
+
+	sourceDb, targetDb, connError := connectToDatabases()
+	if connError != nil {
+		log.Fatal("Connection error: ", connError)
+	}
+
+	defer sourceDb.Close()
+	defer targetDb.Close()
+
+	for _, job := range migrationJobs {
+		log.Infof("Processing job: %+v", job)
+		processMigrationJob(sourceDb, targetDb, job)
+	}
 
 	log.Info("Migration completed successfully!")
 }

cmd/go_migrate/mssql-transform.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"encoding/binary"
+)
+
+func mssqlUuidToBigEndian(mssqlUuid []byte) []byte {
+	if len(mssqlUuid) != 16 {
+		return mssqlUuid
+	}
+	pgUuid := make([]byte, 16)
+	pgUuid[0], pgUuid[1], pgUuid[2], pgUuid[3] = mssqlUuid[3], mssqlUuid[2], mssqlUuid[1], mssqlUuid[0]
+	pgUuid[4], pgUuid[5] = mssqlUuid[5], mssqlUuid[4]
+	pgUuid[6], pgUuid[7] = mssqlUuid[7], mssqlUuid[6]
+	copy(pgUuid[8:], mssqlUuid[8:])
+
+	return pgUuid
+}
+
+const sridFlag = 0x20000000
+
+func wkbToEwkbWithSrid(geometry []byte, srid int) []byte {
+	if len(geometry) < 5 {
+		return geometry
+	}
+
+	var byteOrder binary.ByteOrder
+	if geometry[0] == 0 {
+		byteOrder = binary.BigEndian
+	} else {
+		byteOrder = binary.LittleEndian
+	}
+
+	wkbType := byteOrder.Uint32(geometry[1:5])
+	if wkbType&sridFlag != 0 {
+		return geometry
+	}
+
+	ewkbType := wkbType | sridFlag
+
+	result := make([]byte, len(geometry)+4)
+
+	result[0] = geometry[0]
+
+	byteOrder.PutUint32(result[1:5], ewkbType)
+
+	byteOrder.PutUint32(result[5:9], uint32(srid))
+
+	copy(result[9:], geometry[5:])
+
+	return result
+}

cmd/go_migrate/process.go

@@ -0,0 +1,111 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"sync"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	_ "github.com/microsoft/go-mssqldb"
+	log "github.com/sirupsen/logrus"
+)
+
+func processMigrationJob(sourceDb *sql.DB, targetDb *pgxpool.Pool, job MigrationJob) {
+	sourceColTypes, targetColTypes, err := GetColumnTypes(sourceDb, targetDb, job)
+	if err != nil {
+		log.Fatal("Unexpected error: ", err)
+	}
+
+	logColumnTypes(sourceColTypes, "Source col types")
+	logColumnTypes(targetColTypes, "Target col types")
+
+	chRowsExtract := make(chan []UnknownRowValues, QueueSize)
+	chRowsTransform := make(chan []UnknownRowValues)
+	mssqlContext := context.Background()
+
+	go func() {
+		if err := extractFromMssql(mssqlContext, job, sourceColTypes, ChunkSize, sourceDb, chRowsExtract); err != nil {
+			log.Error("Unexpected error extrating data from mssql: ", err)
+		}
+		close(chRowsExtract)
+	}()
+
+	go func() {
+		transformRowsMssql(sourceColTypes, chRowsExtract, chRowsTransform)
+		close(chRowsTransform)
+	}()
+
+	var wgFakeLoaders sync.WaitGroup
+
+	wgFakeLoaders.Go(func() {
+		fakeLoader(job, sourceColTypes, chRowsTransform)
+	})
+
+	chRowsExtractPostgres := make(chan []UnknownRowValues, QueueSize)
+	postgresContext := context.Background()
+
+	go func() {
+		if err := extractFromPostgres(postgresContext, job, sourceColTypes, ChunkSize, targetDb, chRowsExtractPostgres); err != nil {
+			log.Error("Unexpected error extrating data from postgres: ", err)
+		}
+		close(chRowsExtractPostgres)
+	}()
+
+	wgFakeLoaders.Go(func() {
+		fakeLoader(job, targetColTypes, chRowsExtractPostgres)
+	})
+
+	wgFakeLoaders.Wait()
+}
+
+func logColumnTypes(columnTypes []ColumnType, label string) {
+	log.Info(label)
+
+	for _, col := range columnTypes {
+		log.Infof("%+v", col)
+	}
+}
+
+func transformRowsMssql(columns []ColumnType, in <-chan []UnknownRowValues, out chan<- []UnknownRowValues) {
+	for rows := range in {
+		log.Debugf("Chunk received, transforming...")
+
+		for _, rowValues := range rows {
+			for i, col := range columns {
+				value := rowValues[i]
+				if col.SystemType() == "uniqueidentifier" {
+					if b, ok := value.([]byte); ok {
+						rowValues[i] = mssqlUuidToBigEndian(b)
+					}
+				} else if col.SystemType() == "geometry" || col.SystemType() == "geography" {
+					if b, ok := value.([]byte); ok {
+						rowValues[i] = wkbToEwkbWithSrid(b, 4326)
+					}
+				}
+			}
+		}
+
+		out <- rows
+	}
+}
+
+func logSampleRow(job MigrationJob, columns []ColumnType, rowValues UnknownRowValues, tag string) {
+	log.Infof("[%s.%s] Sample row: (%s)", job.Schema, job.Table, tag)
+	for i, col := range columns {
+		log.Infof("%s (%T): %v", col.Name(), rowValues[i], rowValues[i])
+	}
+}
+
+func fakeLoader(job MigrationJob, columns []ColumnType, in <-chan []UnknownRowValues) {
+	for rows := range in {
+		log.Debugf("Chunk received, loading data into...")
+
+		for i, rowValues := range rows {
+			if i%100 == 0 {
+				logSampleRow(job, columns, rowValues, fmt.Sprintf("row %d", i))
+			}
+		}
+	}
+}