From c5987fa7f8fd3718823adc3ccb853f2322d4b01d Mon Sep 17 00:00:00 2001 From: Kylesoda <249518290+kylesoda@users.noreply.github.com> Date: Wed, 8 Apr 2026 11:00:00 -0500 Subject: [PATCH] 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 | 60 ++++++ cmd/go_migrate/colum-type.go | 44 ++++ cmd/go_migrate/connect.go | 77 +++++++ cmd/go_migrate/extractor.go | 88 ++++++++ cmd/go_migrate/inspect-columns.go | 279 ++++++++++++++++++++++++++ cmd/go_migrate/main.go | 34 ++++ cmd/go_migrate/mssql-transform.go | 52 +++++ cmd/go_migrate/process.go | 111 ++++++++++ 8 files changed, 745 insertions(+) create mode 100644 cmd/go_migrate/build-extract-query.go create mode 100644 cmd/go_migrate/colum-type.go create mode 100644 cmd/go_migrate/connect.go create mode 100644 cmd/go_migrate/extractor.go create mode 100644 cmd/go_migrate/inspect-columns.go create mode 100644 cmd/go_migrate/mssql-transform.go create mode 100644 cmd/go_migrate/process.go diff --git a/cmd/go_migrate/build-extract-query.go b/cmd/go_migrate/build-extract-query.go new file mode 100644 index 0000000..b2a8c4c --- /dev/null +++ b/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) +} diff --git a/cmd/go_migrate/colum-type.go b/cmd/go_migrate/colum-type.go new file mode 100644 index 0000000..cfd76a4 --- /dev/null +++ b/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 +} diff --git a/cmd/go_migrate/connect.go b/cmd/go_migrate/connect.go new file mode 100644 index 0000000..40e417f --- /dev/null +++ b/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 +} diff --git a/cmd/go_migrate/extractor.go b/cmd/go_migrate/extractor.go new file mode 100644 index 0000000..9cf34e4 --- /dev/null +++ b/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 +} diff --git a/cmd/go_migrate/inspect-columns.go b/cmd/go_migrate/inspect-columns.go new file mode 100644 index 0000000..b97043c --- /dev/null +++ b/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 +} diff --git a/cmd/go_migrate/main.go b/cmd/go_migrate/main.go index 6b76845..fe7e579 100644 --- a/cmd/go_migrate/main.go +++ b/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!") } diff --git a/cmd/go_migrate/mssql-transform.go b/cmd/go_migrate/mssql-transform.go new file mode 100644 index 0000000..9f73468 --- /dev/null +++ b/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 +} diff --git a/cmd/go_migrate/process.go b/cmd/go_migrate/process.go new file mode 100644 index 0000000..5f49d47 --- /dev/null +++ b/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)) + } + } + } +}